如何使用 AWS java SDK 使用 AWS 粘合作业生成自动脚本

Posted 2023-02-16

【中文标题】如何使用 AWS java SDK 使用 AWS 粘合作业生成自动脚本

【英文标题】：How to do auto script generation with AWS glue job with AWS java SDK

【发布时间】：2020-05-26 18:50:25

【问题描述】：

我正在使用 Java Sdk 创建粘合作业。它只有两个必需的参数 Command 和 Glue 版本。 但我需要使用自动脚本生成来创建工作。正如我们可以从控制台做的那样，我们添加数据源、AWS Glue 生成的建议脚本、转换类型、数据目标、架构 n 全部。如何使用 java sdk 甚至使用 aws glue api 将这些参数添加到粘合作业。

           CreateJobRequest req = new CreateJobRequest();
            req.setName("TestJob2");
            req.setRole("GlueS3Role");
            req.setGlueVersion("1.0");
            JobCommand command = new JobCommand();
            command.setName("glueetl");
            command.setPythonVersion("3");
            **// S3 location need not to be given, as script code is auto generated by AWS glue
           command.setScriptLocation(S3ScriptLocation);**
            req.setCommand(command);

            AWSGlue glueClient = AWSGlueClientBuilder.standard()
                                 .withRegion(Regions.US_EAST_1)
                                 .withCredentials(new AWSStaticCredentialsProvider(creds))
                                 .build();

            glueClient.createJob(req);
        

【参考方案1】：

我希望 AWS Glue 客户端和触发作业的逻辑的实施将有助于以同样的方式实现胶水作业的自动生成。

胶水客户端：

public GlueClient createClient() 
        return GlueClient.builder()
                .region(Region.of(regionName))
                .credentialsProvider(ProfileCredentialsProvider.create(profileName)).build();

胶水作业运行器：

public static String runGlueJob(GlueClient glueClient, String jobName, Map<String, String> glueArguments) 

        StartJobRunResponse response = glueClient.startJobRun(StartJobRunRequest.builder().jobName(jobName).arguments(glueArguments).build());
        String jobId = response.jobRunId();
        logger.info("JobId: " + jobId);

        return jobId;

    

接下来我们可以创建一个新的 AWS Glue 作业定义：

CreateJobResult jobResult = glueClient.createJob(CreateJobRequest.builder()
                .command(JobCommand.builder().pythonVersion("").scriptLocation("").name("").build())
                .defaultArguments()
                .description()
                .glueVersion()
                .logUri()
                .name()
                .numberOfWorkers()
                .role()
                .tags()
                .build());            

然后触发作业

https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/glue/AWSGlueClient.html

【参考方案2】：

您正在寻找的是createScript(CreateScriptRequest request) - AWSJavaSDK AWSGlueClient 类中的一个函数

很遗憾，当前版本的 AWS Glue SDK 不包含用于生成 ETL 脚本的简单功能。在创建作业功能中生成 ETL 脚本时，AWS Glue 控制台本身会在后台执行多项操作（您可以通过查看浏览器的网络选项卡来查看这一点）。

使用“DAG”来模仿这个

您需要收集CodeGenNode 和CodeGenEdge 并将它们添加到您的CreateScriptRequest 中

.WithDagNodes(Collection<CodeGenNode> collection)

&

.WithDagEdges(Collection<CodeGenEdge> collection)

我建议您首先在 AWS 控制台中生成一个 ETL 脚本，并将该结果与 "Generate Scala Code" example 中的信息交叉引用（此链接是为了让您更好地理解“DAG”）

我最终明确地构建了这个 DAG 结构。这是我的解决方案的 sn-p：

    var dagNodes = new ArrayList<CodeGenNode>();
    var dagEdges = new ArrayList<CodeGenEdge>();

    //datasource
    dagEdges.add(new CodeGenEdge().withSource(dataSourceName).withTarget(applyMappingName));
    ArrayList<CodeGenNodeArg> dataSourceArgs = new ArrayList<CodeGenNodeArg>();
    dataSourceArgs.add(new CodeGenNodeArg().withName("database").withValue(String.format("\"%s\"", databaseName)));
    dataSourceArgs.add(new CodeGenNodeArg().withName("table_name").withValue(String.format("\"%s\"", tableName)));
    dataSourceArgs.add(new CodeGenNodeArg().withName("transformation_ctx").withValue(String.format("\"%s\"", dataSourceName)));
    dagNodes.add(new CodeGenNode().withId(dataSourceName).withNodeType("DataSource").withArgs(dataSourceArgs));

... //can build out many 'operations' - datasource, applymapping, selectfields, resolvechoice, datasink

    var createScriptRequest = new CreateScriptRequest()
        .withDagEdges(dagEdges)
        .withDagNodes(dagNodes)
        .withLanguage(Language.PYTHON);

    awsGlueClient.createScript(createScriptRequest)

然后只需使用AmazonS3 将此结果上传到 S3 并将此路径用于“setScriptLocation”

PutObjectResult putObject(String bucketName, String key, String content)