Spark Job в конвейере данных AWS с учетными данными хранилища параметров SSM

Question

Я пытаюсь запустить задание Spark в AWS Data Pipeline и хочу передать учетные данные в качестве параметра приложению Spark и получить эти учетные данные из хранилища параметров SSM.

Это то, что у меня есть для моегошаг в EmrActivity в моем конвейере.

command-runner.jar,spark-submit,--deploy-mode,cluster,--master,yarn,--class,MyClass,s3://MyFatJar.jar,$SALT,$PASS

Когда я запускаю aws emr локально, я использую переменные локальной среды $ SALT и $ PASS. В конвейере данных я хочу получить их из хранилища параметров.

У меня есть этот код, чтобы получить соль из хранилища параметров, но я не знаю, как запустить это в конвейере и передать его искре. step.

SALT=$(aws ssm get-parameter --name salt --with-decryption --query 'Parameter.Value' --output text)

Я попытался вставить этот код bash в команду предварительного шага EmrActivity, я попытался создать ShellCommandActivity и сделать emr зависимым от этого, я также попытался поместить его в шаг с помощью команды runner. баночка, Баш. Я не смог заставить ни одного из них работать.

Вот определение конвейера моей текущей попытки:

{
  "objects": [
    {
      "schedule": {
        "ref": "DefaultSchedule"
      },
      "name": "DefaultShellCommandActivity2",
      "runsOn": {
        "ref": "EmrClusterObj"
      },
      "id": "ShellCommandActivityId_bWuUG",
      "type": "ShellCommandActivity",
      "command": "PASS=$(aws ssm get-parameter --name pass --with-decryption --query 'Parameter.Value' --output text)"
    },
    {
      "name": "EmrActivityObj",
      "step": "#{myEmrStep}",
      "runsOn": {
        "ref": "EmrClusterObj"
      },
      "id": "EmrActivityObj",
      "type": "EmrActivity"
    },
    {
      "period": "1 days",
      "name": "Every 1 day",
      "id": "DefaultSchedule",
      "type": "Schedule",
      "startAt": "FIRST_ACTIVATION_DATE_TIME"
    },
    {
      "taskInstanceType": "#{myTaskInstanceType}",
      "coreInstanceCount": "#{myCoreInstanceCount}",
      "masterInstanceType": "#{myMasterInstanceType}",
      "releaseLabel": "#{myEMRReleaseLabel}",
      "type": "EmrCluster",
      "terminateAfter": "50 Minutes",
      "bootstrapAction": "#{myBootstrapAction}",
      "taskInstanceCount": "#{myTaskInstanceCount}",
      "name": "EmrClusterObj",
      "coreInstanceType": "#{myCoreInstanceType}",
      "keyPair": "#{myEC2KeyPair}",
      "id": "EmrClusterObj",
      "applications": "Spark"
    },
    {
      "failureAndRerunMode": "CASCADE",
      "schedule": {
        "ref": "DefaultSchedule"
      },
      "resourceRole": "DataPipelineDefaultResourceRole",
      "role": "DataPipelineDefaultRole",
      "pipelineLogUri": "s3://data-pipeline-logs",
      "scheduleType": "cron",
      "name": "Default",
      "id": "Default"
    },
    {
      "schedule": {
        "ref": "DefaultSchedule"
      },
      "name": "DefaultShellCommandActivity1",
      "runsOn": {
        "ref": "EmrClusterObj"
      },
      "id": "ShellCommandActivityId_ew8w5",
      "type": "ShellCommandActivity",
      "command": "SALT=$(aws ssm get-parameter --name salt --with-decryption --query 'Parameter.Value' --output text)"
    }
  ],
  "parameters": [
    {
      "helpText": "An existing EC2 key pair to SSH into the master node of the EMR cluster as the user \"hadoop\".",
      "description": "EC2 key pair",
      "optional": "true",
      "id": "myEC2KeyPair",
      "type": "String"
    },
    {
      "helpLink": "https://docs.aws.amazon.com/console/datapipeline/emrsteps",
      "watermark": "s3://myBucket/myPath/myStep.jar,firstArg,secondArg",
      "helpText": "A step is a unit of work you submit to the cluster. You can specify one or more steps",
      "description": "EMR step(s)",
      "isArray": "true",
      "id": "myEmrStep",
      "type": "String"
    },
    {
      "helpText": "Task instances run Hadoop tasks.",
      "description": "Task node instance type",
      "optional": "true",
      "id": "myTaskInstanceType",
      "type": "String"
    },
    {
      "default": "m1.medium",
      "helpText": "Core instances run Hadoop tasks and store data using the Hadoop Distributed File System (HDFS).",
      "description": "Core node instance type",
      "id": "myCoreInstanceType",
      "type": "String"
    },
    {
      "default": "emr-5.13.0",
      "helpText": "Determines the base configuration of the instances in your cluster, including the Hadoop version.",
      "description": "EMR Release Label",
      "id": "myEMRReleaseLabel",
      "type": "String"
    },
    {
      "default": "2",
      "description": "Core node instance count",
      "id": "myCoreInstanceCount",
      "type": "Integer"
    },
    {
      "description": "Task node instance count",
      "optional": "true",
      "id": "myTaskInstanceCount",
      "type": "Integer"
    },
    {
      "helpLink": "https://docs.aws.amazon.com/console/datapipeline/emr_bootstrap_actions",
      "helpText": "Bootstrap actions are scripts that are executed during setup before Hadoop starts on every cluster node.",
      "description": "Bootstrap action(s)",
      "isArray": "true",
      "optional": "true",
      "id": "myBootstrapAction",
      "type": "String"
    },
    {
      "default": "m1.medium",
      "helpText": "The Master instance assigns Hadoop tasks to core and task nodes, and monitors their status.",
      "description": "Master node instance type",
      "id": "myMasterInstanceType",
      "type": "String"
    }
  ],
  "values": {
    "myMasterInstanceType": "m3.2xlarge",
    "myEMRReleaseLabel": "emr-5.11.0",
    "myEmrStep": "command-runner.jar,spark-submit,--deploy-mode,cluster,--master,yarn,--class,MyClass,s3://MyFatJar.jar,$PASS,$SALT",
    "myCoreInstanceCount": "4",
    "myCoreInstanceType": "m3.2xlarge"
  }
}