支援串流連接器

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

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

Amazon Kinesis Data Streams 連接器

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

以下是如何使用 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無伺服器應用程式，並使用VPC端點來允許私人存取。或使用NAT閘道取得公開存取權。如需詳細資訊，請參閱設定VPC存取權。您也必須確定您的工作執行階段角色具有必要的讀取和寫入權限，才能存取所需的資料串流。若要進一步了解如何設定 Job 務執行時期角色，請參閱 Amazon EMR 無伺服器的任務執行階段角色。如需所有必要權限的完整清單，請參閱上的spark-sql-kinesis-connector 頁面 GitHub。

阿帕奇卡夫卡連接器

用於星火結構化流阿帕奇卡夫卡連接器是來自星火社區的開源連接器，並在 Maven 存儲庫中可用。此連接器可協助 Spark 結構化串流應用程式讀取資料，並將資料寫入自我管理的 Apache Kafka，以及適用於 Apache Kafka 的 Amazon 受管串流。如需有關連接器的詳細資訊，請參閱 Apache Spark 說明文件中的結構化串流 + 卡夫卡整合指南。

下列範例示範如何在工作執行要求中包含 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無伺服器發行版本和對應的 Spark 版本。要查找正確的卡夫卡版本，請參閱結構化流 + 卡夫卡集成指南。

若要透過IAM身份驗證使用適用於 Apache Kafka 的 Amazon 受管串流，您必須包含另一個相依性，以使 Kafka 連接器能夠透過連接到 Amazon。MSK IAM如需詳細資訊，請參閱（詳見）的aws-msk-iam-auth 存放庫 GitHub。您也必須確定工作執行階段角色具有必要的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>"
    }
}'

若要使用來自 Amazon 的 Kafka 連接器和IAM身份驗證程式庫，MSK您必須設定具有存取權的EMR無伺服器應用程式。VPC您的子網必須具有 Internet 訪問權限，並使用NAT網關訪問 Maven 依賴關係。如需詳細資訊，請參閱設定VPC存取權。子網路必須具有網路連線能力才能存取 Kafka 叢集。無論您的 Kafka 叢集是自我管理還是使用適用於 Apache Kafka 的 Amazon 受管串流，都是如此。