在 Amazon Glue 中流式处理 ETL 作业

• Avro 串流源的注释和限制

• 将 grok 模式应用于串流源

为 Amazon Kinesis Data Streams 设置串流 ETL 属性（Amazon Glue API 或 Amazon CLI）

• 要为同一账户中的 Kinesis 源设置串流 ETL 属性，请在 CreateTable API 操作或 create_table CLI 命令的 StorageDescriptor 结构中指定 streamName 和 endpointUrl 参数。

  "StorageDescriptor": {
  	"Parameters": {
  		"typeOfData": "kinesis",
  		"streamName": "sample-stream",
  		"endpointUrl": "https://kinesis.us-east-1.amazonaws.com"
  	}
  	...
  }
  					

  或者，指定 streamARN。

  "StorageDescriptor": {
  	"Parameters": {
  		"typeOfData": "kinesis",
  		"streamARN": "arn:aws:kinesis:us-east-1:123456789:stream/sample-stream"
  	}
  	...
  }

• 要为其他账户中的 Kinesis 源设置串流 ETL 属性，请在 CreateTable API 操作或 create_table CLI 命令的 StorageDescriptor 结构中指定 streamARN、awsSTSRoleARN 和 awsSTSSessionName（可选）参数。

  "StorageDescriptor": {
  	"Parameters": {
  		"typeOfData": "kinesis",
  		"streamARN": "arn:aws:kinesis:us-east-1:123456789:stream/sample-stream",
  		"awsSTSRoleARN": "arn:aws:iam::123456789:role/sample-assume-role-arn",
  		"awsSTSSessionName": "optional-session"
  	}
  	...
  }

• 启用架构检测后，Avro 架构必须包含在负载中。关闭时，负载应仅包含数据。

• 某些 Avro 数据类型在动态帧中不受支持。在使用 Amazon Glue 控制台的创建表向导中的 Define a schema (定义架构) 页面时，您无法指定这些数据类型。在架构检测期间，Avro 架构中不受支持的类型将转换为受支持的类型，如下所示：

  • EnumType => StringType

  • FixedType => BinaryType

  • UnionType => StructType

• 如果使用控制台中的 Define a schema (定义架构) 页面，则架构的隐含根元素类型为 record。如果你想要一个除 record 以外的根元素类型，例如 array 或者 map，则不能使用 Define a schema (定义架构) 页面来指定架构。相反，您必须跳过该页并将架构指定为表属性或在 ETL 脚本中指定。

  • 要在表属性中指定架构，请完成创建表向导，编辑表详细信息，并在 Table properties (表属性) 下添加新的键值对。使用密钥 avroSchema，然后为值输入架构 JSON 对象，如以下屏幕截图所示。

    

  • 要在 ETL 脚本中指定架构，请修改 datasource0 任务语句并将 avroSchema 密钥添加到 additional_options 参数，如以下 Python 和 Scala 示例所示。

    Python

    SCHEMA_STRING = ‘{"type":"array","items":"string"}’
    datasource0 = glueContext.create_data_frame.from_catalog(database = "database", table_name = "table_name", transformation_ctx = "datasource0", additional_options = {"startingPosition": "TRIM_HORIZON", "inferSchema": "false", "avroSchema": SCHEMA_STRING})
    

    Scala

    val SCHEMA_STRING = """{"type":"array","items":"string"}"""
    val datasource0 = glueContext.getCatalogSource(database = "database", tableName = "table_name", redshiftTmpDir = "", transformationContext = "datasource0", additionalOptions = JsonOptions(s"""{"startingPosition": "TRIM_HORIZON", "inferSchema": "false", "avroSchema":"$SCHEMA_STRING"}""")).getDataFrame()
    

• 自动解压 Amazon Glue 流式传输 ETL 任务仅适用于受支持的压缩类型。另请注意以下几点：

  • Framed Snappy 是指适用于 Snappy 的帧格式。

  • Deflate 在 Glue 版本 3.0 而不是 Glue 版本 2.0 中受支持。

• 使用架构检测时，无法执行串流数据联接。

• Amazon Glue 流式传输 ETL 任务不支持对具有 Avro 格式的 Amazon Glue 架构注册表使用 Union 数据类型。

• 您的 ETL 脚本可以使用 Amazon Glue 的内置转换和 Apache Spark Structured Streaming 的原生转换。有关更多信息，请参阅 Apache Spark 网站或 Amazon Glue PySpark 转换参考 上的串流 DataFrame/数据集的操作。

• Amazon Glue 串流 ETL 任务使用检查点来跟踪已读取的数据。因此，停止并重新启动的任务将从流中停止的位置开始。如果要重新处理数据，您可以删除脚本中引用的检查点文件夹。

• 不支持任务书签。

• 要在作业中使用 Kinesis Data Streams 的增强型扇出功能，请参阅 在 Kinesis 流作业中使用增强型扇出功能。

• 如果您使用在 Amazon Glue 架构注册表中创建的数据目录表，则当新的架构版本可用时，要反映该新架构，您需要执行以下操作：

  1. 停止与该表关联的任务。

  2. 更新数据目录表的架构。

  3. 重新启动与该表关联的任务。