EMR - Airflow 运行 scala jar 文件airflow.exceptions.AirflowException

Question

我正在尝试使用 emr 从 AIRFLOW 运行 scala jar 文件，该 jar 文件旨在读取 mssql-jdbc 和 postgresql。

通过气流，我可以创建集群

我的 SPARK_STEPS 看起来像

SPARK_STEPS = [
    {
        'Name': 'Trigger_Source_Target',
        'ActionOnFailure': 'CONTINUE',
        'HadoopJarStep': {
            'Jar': 'command-runner.jar',
            'Args': ['spark-submit', 
                     '--master', 'yarn', 
                     '--jars', '/mnt/MyScalaImport.jar', 
                     '--class', 'org.classname', 
                     's3://path/SNAPSHOT.jar', 
                     'SQL_Pwd', 'PostgreSQL_PWD', 'loadtype'],
        }
    }
]

在此之后我定义了 JOB_FLOW_OVERRIDES -

JOB_FLOW_OVERRIDES = {
    "Name": "pfdt-cluster-airflow",
    "LogUri": "s3://path/elasticmapreduce/",
    "ReleaseLabel": "emr-6.4.0",
    "Applications": [
        {"Name": "Spark"},
    ],
    "Instances": {
        "InstanceGroups": [
            {
                "Name": "Master nodes",
                "Market": "ON_DEMAND",
                "InstanceRole": "MASTER",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 1,
            }
        ],
        "KeepJobFlowAliveWhenNoSteps": True,
        "TerminationProtected": False,
        'Ec2KeyName': 'pem_file_name',
        "Ec2SubnetId": "subnet-123"
    },
    'BootstrapActions': [
        {
            'Name': 'import custom Jars',
            'ScriptBootstrapAction': {
                'Path': 's3://path/subpath/copytoolsjar.sh',
                'Args': []
            }
        }
    ],
    'Configurations': [
      {
        'Classification': 'spark-defaults',
        'Properties': {
          'spark.jars': 's3://jar_path/mssql-jdbc-8.4.1.jre8.jar'
         }
      }
    ],
    "VisibleToAllUsers": True,
    "JobFlowRole": "EMR_EC2_DefaultRole",
    "ServiceRole": "EMR_DefaultRole",
    "Tags": [
        {"Key": "Environment", "Value": "Development"},
    ],
}

将 scala .jar 文件从 S3 复制到本地气流-我有一个可以完成工作的shell脚本：Path- s3://path/subpath/copytoolsjar.sh

aws s3 cp s3://path/SNAPSHOT.jar /mnt/MyScalaImport.jar

在触发气流时 - 它在节点 watch_step

我收到的错误是- stdout.gz => stderr.gz =>

22/04/08 13:38:23 INFO CodeGenerator：在 25.5907 毫秒内生成代码 线程“主”java.sql.SQLException 中出现异常：没有合适的驱动程序 在 java.sql.DriverManager.getDriver(DriverManager.java:315) 在 org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.$anonfun$driverClass$2(JDBCOptions.scala:108) 在 scala.Option.getOrElse(Option.scala:189) 在org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions。（JDBCOptions.scala：108） 在 org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.(JDBCOptions.scala:38)

如何解决此问题 - 我的罐子在-

s3://路径/子路径/mssql-jdbc-8.4.1.jre8.jar

s3://路径/子路径/postgresql-42.2.24.jar

Answer 1

要上传JAR文件（MSSQL-JDBC-8.4.1.JRE8.JAR，POSTGRESQL-42.2.24.JAR）到

Bootstrap step-
'bootstrapactions'：[{'name'：'import custom jars'，'scriptbootStrapaction'：{'path'：'s3：'s3：//path/subpath/copytoolsjar.sh''，'args'：[]}}}} ]

在copytoolsjar.sh文件中写下命令as-
AWS S3 CP CP S3：//path/snapshot.jar/mnt/myscalaimport.jar＆amp;＆amp;＆amp; bash -c“ sudo aws s3 cp s3：//path/subpath/mssql-jdbc-8.4.1.jre8.jre8.jar/iusr/iusr/ib/lib/park/aspar/jars/”＆amp;＆amp;＆amp; bash -c“ sudo aws s3 cp s3：//path/subpath/postgresql-42.2.24.2.24.jar/usr/lib/lib/spark/jars/'
工作将完成