如何配置粘合书签以与 scala 代码一起使用？

考虑 Scala 代码：

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.{GlueArgParser, Job, JsonOptions}
import org.apache.spark.SparkContext

import scala.collection.JavaConverters.mapAsJavaMapConverter

object MyGlueJob {

  def main(sysArgs: Array[String]) {
    val spark: SparkContext = SparkContext.getOrCreate()
    val glueContext: GlueContext = new GlueContext(spark)

    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)

    val input = glueContext
      .getCatalogSource(database = "my_data_base", tableName = "my_json_gz_partition_table")
      .getDynamicFrame()

    val processed = input.applyMapping(
      Seq(
        ("id",                                        "string", "id", "string"),
        ("my_date",                                   "string", "my_date", "string")
      ))
    glueContext.getSinkWithFormat(
      connectionType = "s3",
      options = JsonOptions(Map("path" -> "s3://my_path", "partitionKeys" -> List("my_date"))),
      format = "orc", transformationContext = ""
    ).writeDynamicFrame(processed)
    Job.commit
  }
}

输入是使用 gzip 压缩的分区 json 文件，该文件按日期列分区。一切works- 数据以json格式读取，以orc格式写入。

但是，当尝试使用相同的数据运行作业时，它会再次读取并写入重复的数据。该作业中启用了书签。方法Job.init and Job.commit被调用。怎么了？

UPDATED

我添加了一个transformationContext参数为getCatalogSource and getSinkWithFormat:

        val input = glueContext
      .getCatalogSource(database = "my_data_base", tableName = "my_json_gz_partition_table", transformationContext = "transformationContext1")
      .getDynamicFrame()

and:

    glueContext.getSinkWithFormat(
      connectionType = "s3",
      options = JsonOptions(Map("path" -> "s3://my_path", "partitionKeys" -> List("my_date"))),
      format = "orc", transformationContext = "transformationContext2"
    ).writeDynamicFrame(processed)

现在魔法就这样“起作用”了：

1. 第一次运行 - 好的
2. 第二次运行（使用相同的数据或相同的数据和新的数据） - 它失败并出现错误（稍后）

第二次（及后续）运行后再次发生错误。 还有留言Skipping Partition {"my_date": "2017-10-10"}出现在日志中。

ERROR ApplicationMaster: User class threw exception: org.apache.spark.sql.AnalysisException: Partition column my_date not found in schema StructType(); org.apache.spark.sql.AnalysisException: Partition column my_date not found in schema StructType();
at org.apache.spark.sql.execution.datasources.PartitioningUtils$$anonfun$partitionColumnsSchema$1$$anonfun$apply$11.apply(PartitioningUtils.scala:439)
at org.apache.spark.sql.execution.datasources.PartitioningUtils$$anonfun$partitionColumnsSchema$1$$anonfun$apply$11.apply(PartitioningUtils.scala:439)
at scala.Option.getOrElse(Option.scala:121)
at org.apache.spark.sql.execution.datasources.PartitioningUtils$$anonfun$partitionColumnsSchema$1.apply(PartitioningUtils.scala:438)
at org.apache.spark.sql.execution.datasources.PartitioningUtils$$anonfun$partitionColumnsSchema$1.apply(PartitioningUtils.scala:437)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.immutable.List.foreach(List.scala:381)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.immutable.List.map(List.scala:285)
at org.apache.spark.sql.execution.datasources.PartitioningUtils$.partitionColumnsSchema(PartitioningUtils.scala:437)
at org.apache.spark.sql.execution.datasources.PartitioningUtils$.validatePartitionColumn(PartitioningUtils.scala:420)
at org.apache.spark.sql.execution.datasources.DataSource.write(DataSource.scala:443)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:215)
at com.amazonaws.services.glue.SparkSQLDataSink.writeDynamicFrame(DataSink.scala:123)
at MobileArcToRaw$.main(script_2018-01-18-08-14-38.scala:99)

胶水书签究竟是怎么回事？？？哦

您是否尝试过设置transformationContext源和接收器的值是否相同？它们当前在上次更新中设置为不同的值。

transformationContext = "transformationContext1"

and

transformationContext = "transformationContext2"

我在使用胶水和书签时也遇到过这个问题。我正在尝试执行类似的任务，其中读取按年、月和日分区的分区 JSON 文件，每天都会有新文件到达。我的工作运行一个转换来提取数据的子集，然后放入 S3 上的分区 Parquet 文件中。

我使用的是 Python，因此 DynamicFrame 的初始实例化如下所示：

dyf = glue_context.create_dynamic_frame.from_catalog(database="dev-db", table_name="raw", transformation_ctx="raw")

最后像这样接收到 S3：

glue_context.write_dynamic_frame.from_options( frame=select_out, connection_type='s3', connection_options={'path': output_dir, 'partitionKeys': ['year', 'month', 'day']}, format='parquet', transformation_ctx="dev-transactions" )

最初，我运行了该作业，并在启用书签的情况下正确生成了 Parquet。然后我添加了新一天的数据，更新了输入表上的分区并重新运行。第二个作业将失败并出现如下错误：

pyspark.sql.utils.AnalysisException: u"cannot resolve 'year' given input columns: [];;\n'Project ['year, 'month, 'day, 'data']

改变transformation_ctx是相同的（dev-transactions在我的例子中）使该过程能够正常工作，仅处理增量分区并为新分区生成 Parquet。

关于一般书签以及如何使用转换上下文变量的文档非常稀疏。

Python 文档只是说：（https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-glue-context.html https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-glue-context.html):

conversion_ctx – 要使用的转换上下文（可选）。

Scala 文档说（https://docs.aws.amazon.com/glue/latest/dg/glue-etl-scala-apis-glue-gluecontext.html https://docs.aws.amazon.com/glue/latest/dg/glue-etl-scala-apis-glue-gluecontext.html):

conversionContext — 与作业书签使用的接收器关联的转换上下文。默认设置为空。

由于文档解释得不好，我所能观察到的最好情况是，转换上下文用于在已处理的源数据和接收器数据之间形成链接，并且定义不同的上下文会阻止书签按预期工作。