支援的串流連接器

串流連接器有助於從串流來源讀取資料，也可以將資料寫入串流接收器。

以下是支援的串流連接器：

Amazon Kinesis Data Streams 連接器

適用於 Apache Spark 的 Amazon Kinesis Data Streams 連接器可建置串流應用程式和管道，以取用來自 的資料，並將資料寫入 Amazon Kinesis Data Streams。連接器支援增強的廣發耗用，專用讀取輸送量速率高達每個碎片 2MB/秒。根據預設，Amazon EMR Serverless 7.1.0 及更高版本包含 連接器，因此您不需要建置或下載任何其他套件。如需連接器的詳細資訊，請參閱 GitHub 上的 spark-sql-kinesis-connector 頁面。

以下是如何使用 Kinesis Data Streams 連接器相依性啟動任務執行的範例。

aws emr-serverless start-job-run \
--application-id <APPLICATION_ID> \
--execution-role-arn <JOB_EXECUTION_ROLE> \
--mode 'STREAMING' \
--job-driver '{
    "sparkSubmit": {
        "entryPoint": "s3://<Kinesis-streaming-script>",
        "entryPointArguments": ["s3://<DOC-EXAMPLE-BUCKET-OUTPUT>/output"],
        "sparkSubmitParameters": "--conf spark.executor.cores=4
                --conf spark.executor.memory=16g 
                --conf spark.driver.cores=4
                --conf spark.driver.memory=16g 
                --conf spark.executor.instances=3
                --jars /usr/share/aws/kinesis/spark-sql-kinesis/lib/spark-streaming-sql-kinesis-connector.jar"
    }
}'

若要連線至 Kinesis Data Streams，請使用 VPC 存取設定 EMR Serverless 應用程式，並使用 VPC 端點來允許私有存取。 或使用 NAT Gateway 來取得公有存取。如需詳細資訊，請參閱設定 VPC 存取。您還必須確保任務執行期角色具有必要的讀取和寫入許可，以存取所需的資料串流。若要進一步了解如何設定任務執行期角色，請參閱 Amazon EMR Serverless 的任務執行期角色。如需所有必要許可的完整清單，請參閱 GitHub 上的 spark-sql-kinesis-connector 頁面。

Apache Kafka 連接器

適用於 Spark 結構化串流的 Apache Kafka 連接器是 Spark 社群的開放原始碼連接器，可在 Maven 儲存庫中使用。此連接器有助於 Spark 結構化串流應用程式從自我管理的 Apache Kafka 和 Amazon Managed Streaming for Apache Kafka 讀取和寫入資料。如需連接器的詳細資訊，請參閱 Apache Spark 文件中的結構化串流 + Kafka 整合指南。

下列範例示範如何在任務執行請求中包含 Kafka 連接器。

aws emr-serverless start-job-run \
--application-id <APPLICATION_ID> \
--execution-role-arn <JOB_EXECUTION_ROLE> \
--mode 'STREAMING' \
--job-driver '{
    "sparkSubmit": {
        "entryPoint": "s3://<Kafka-streaming-script>",
        "entryPointArguments": ["s3://<DOC-EXAMPLE-BUCKET-OUTPUT>/output"],
        "sparkSubmitParameters": "--conf spark.executor.cores=4
                --conf spark.executor.memory=16g 
                --conf spark.driver.cores=4
                --conf spark.driver.memory=16g 
                --conf spark.executor.instances=3
                --packages org.apache.spark:spark-sql-kafka-0-10_2.12:<KAFKA_CONNECTOR_VERSION>"
    }
}'

Apache Kafka 連接器版本取決於您的 EMR Serverless 發行版本和對應的 Spark 版本。若要尋找正確的 Kafka 版本，請參閱 結構化串流 + Kafka 整合指南。

若要搭配 IAM 身分驗證使用 Amazon Managed Streaming for Apache Kafka，請包含另一個相依性，讓 Kafka 連接器能夠透過 IAM 連線至 Amazon MSK。如需詳細資訊，請參閱 GitHub 上的 aws-msk-iam-auth 儲存庫。您也必須確保任務執行時間角色具有必要的 IAM 許可。下列範例示範如何使用具有 IAM 身分驗證的連接器。

aws emr-serverless start-job-run \
--application-id <APPLICATION_ID> \
--execution-role-arn <JOB_EXECUTION_ROLE> \
--mode 'STREAMING' \
--job-driver '{
    "sparkSubmit": {
        "entryPoint": "s3://<Kafka-streaming-script>",
        "entryPointArguments": ["s3://<DOC-EXAMPLE-BUCKET-OUTPUT>/output"],
        "sparkSubmitParameters": "--conf spark.executor.cores=4
                --conf spark.executor.memory=16g 
                --conf spark.driver.cores=4
                --conf spark.driver.memory=16g 
                --conf spark.executor.instances=3
                --packages org.apache.spark:spark-sql-kafka-0-10_2.12:<KAFKA_CONNECTOR_VERSION>,software.amazon.msk:aws-msk-iam-auth:<MSK_IAM_LIB_VERSION>"
    }
}'

若要使用 Kafka 連接器和來自 Amazon MSK 的 IAM 身分驗證程式庫，請設定具有 VPC 存取的 EMR Serverless 應用程式。您的子網路必須具有網際網路存取權，並使用 NAT Gateway 來存取 Maven 相依性。如需詳細資訊，請參閱設定 VPC 存取。子網路必須具有網路連線才能存取 Kafka 叢集。無論您的 Kafka 叢集是自我管理，或是您使用 Amazon Managed Streaming for Apache Kafka，都是如此。