# 使用流式处理数据源
<a name="edit-jobs-source-streaming"></a>

您可以创建连续运行并使用来自流式处理源的数据的流式处理提取、转换和负载（ETL）任务，例如 Amazon Kinesis Data Streams、Apache Kafka 和 Amazon Managed Streaming for Apache Kafka（Amazon MSK）。

**配置流式处理数据源的属性**

1. 转到新任务或已保存任务的可视化图形编辑器。

1. 在图形中为 Kafka 或 Kinesis Data Streams 选择数据源节点。

1. 选择 **Data source properties (数据源属性)** 选项卡，然后输入以下信息：

------
#### [ Kinesis ]
   + **Kinesis source type**（Kinesis 源类型）：选择选项 **Stream details**（流式传输详细信息）可使用直接访问串流源，或选择 **Data Catalog table**（数据目录表）使用存储在其中的信息。

     如果选择 **Stream details**（流式传输详细信息），请指定以下附加信息。
     + **数据流的位置**：选择流式传输与当前用户关联，还是其他用户关联。
     + **Region**（区域）：选择存在流式传输的 AWS 区域。此信息用于构建用于访问数据流的 ARN。
     + **Stream ARN**（流式传输 ARN）：输入 Kinesis 数据流的 Amazon Resource Name (ARN)。如果流式传输位于当前账户内，则可以从下拉列表中选择流式传输名称。您可以使用搜索字段按名称或 ARN 搜索数据流。
     + **Data format**（数据格式）：从列表中选择数据流使用的格式。

       AWS Glue 会自动从流式处理数据中检测架构。

     如果选择 **Data Catalog table**（数据目录表），请指定以下附加信息。
     + **Database (数据库)**：（可选）在 AWS Glue 数据目录中选择数据库，该数据库包含与流式处理数据源关联的表。您可以使用搜索字段按名称搜索数据库。
     + **Table (表)**：（可选）从列表中选择与源数据关联的表。此表必须已存在于 AWS Glue 数据目录中。您可以使用搜索字段按名称搜索表。
     + **Detect schema (检测架构)**：选择该选项，允许 AWS Glue 从流式处理数据检测架构，而不是使用数据目录表中的架构信息。如果选择 **Stream details**（流式传输详细信息）选项，则会自动启用此选项。
   + **Starting position**（起始位置）：预设情况下，ETL 任务会使用 **Earliest**（最早）选项，这意味着从流式传输中最早的可用记录开启读取数据。您也可以选择 **Latest**（最新），表示 ETL 任务应该从流式传输中最新的记录之后开启读取。
   + **Window size (窗口大小)**：默认情况下，在 100 秒的时段内处理 ETL 任务和写出数据。这可以实现数据的高效处理，并允许对晚于预计时间到达的数据执行聚合。您可以修改此窗口大小以提高及时性或聚合精度。

     AWS Glue 流式处理任务使用检查点而非任务书签来跟踪已读取的数据。
   + **Connection options**（连接选项）：展开此部分以添加键值对，指定其他连接选项。有关您可在此处指定的选项的信息，请参阅*AWS Glue 开发人员指南*中的 ["connectionType": "kinesis"](https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-connect.html#aws-glue-programming-etl-connect-kinesis)。

------
#### [ Kafka ]
   + **Apache Kafka source**（Apache Kafka 源）：选择选项 **Stream details**（流式传输详细信息）可使用直接访问流式处理源，或选择 **Data Catalog table**（数据目录表）使用存储在其中的信息。

     如果选择 **Data Catalog table**（数据目录表），请指定以下附加信息。
     + **Database (数据库)**：（可选）在 AWS Glue 数据目录中选择数据库，该数据库包含与流式处理数据源关联的表。您可以使用搜索字段按名称搜索数据库。
     + **Table (表)**：（可选）从列表中选择与源数据关联的表。此表必须已存在于 AWS Glue 数据目录中。您可以使用搜索字段按名称搜索表。
     + **Detect schema (检测架构)**：选择该选项，允许 AWS Glue 从流式处理数据检测架构，而不是将架构信息存储于数据目录表。如果选择 **Stream details**（流式传输详细信息）选项，则会自动启用此选项。

     如果选择 **Stream details**（流式传输详细信息），请指定以下附加信息。
     + **Connection name**（连接名称）：选择包含 Kafka 数据流的访问和身份验证信息的 AWS Glue 连接。您必须使用与 Kafka 流式处理数据源的连接。如果连接不存在，可以使用 AWS Glue 控制台为 Kafka 数据流创建连接。
     + **Topic name**（主题名称）：输入要从中读取的主题的名称。
     + **Data format**（数据格式）：选择从 Kafka 事件流读取数据时使用的格式。
   + **Starting position**（起始位置）：预设情况下，ETL 任务会使用 **Earliest**（最早）选项，这意味着从流式传输中最早的可用记录开启读取数据。您也可以选择 **Latest**（最新），表示 ETL 任务应该从流式传输中最新的记录之后开启读取。
   + **Window size (窗口大小)**：默认情况下，在 100 秒的时段内处理 ETL 任务和写出数据。这可以实现数据的高效处理，并允许对晚于预计时间到达的数据执行聚合。您可以修改此窗口大小以提高及时性或聚合精度。

     AWS Glue 流式处理任务使用检查点而非任务书签来跟踪已读取的数据。
   + **Connection options**（连接选项）：展开此部分以添加键值对，指定其他连接选项。有关您可在此处指定的选项的信息，请参阅*AWS Glue 开发人员指南*中的 ["connectionType": "kafka"](https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-connect.html#aws-glue-programming-etl-connect-kafka)。

------

**注意**  
流式处理数据源当前不支持数据预览。