步驟一：創(chuàng)建實時數(shù)據(jù)流集群并產生消息

1. 在EMR on ECS頁面，創(chuàng)建包含Kafka服務的實時數(shù)據(jù)流集群，詳情請參見創(chuàng)建集群。

2. 登錄EMR集群的Master節(jié)點，詳情請參見登錄集群。

3. 執(zhí)行以下命令，切換目錄。

   cd /var/log/emr/taihao_exporter

4. 執(zhí)行以下命令，創(chuàng)建Topic。

   # 創(chuàng)建名為taihaometrics的Topic，分區(qū)數(shù)10，副本因子2。
   kafka-topics.sh --partitions 10 --replication-factor 2 --bootstrap-server core-1-1:9092 --topic taihaometrics --create

5. 執(zhí)行以下命令，發(fā)送消息。

   # 使用kafka-console-producer發(fā)送消息到taihaometrics Topic。
   tail -f metrics.log | kafka-console-producer.sh --broker-list core-1-1:9092 --topic taihaometrics

步驟二：新增網(wǎng)絡連接

1. 進入網(wǎng)絡連接頁面。

   1. 在EMR控制臺的左側導航欄，選擇EMR Serverless > Spark。

   2. 在Spark頁面，單擊目標工作空間名稱。

   3. 在EMR Serverless Spark頁面，單擊左側導航欄中的網(wǎng)絡連接。

2. 在網(wǎng)絡連接頁面，單擊新增網(wǎng)絡連接。

3. 在新增網(wǎng)絡連接對話框中，配置以下信息，單擊確定。

   參數(shù)	說明
   連接名稱	輸入新增連接的名稱。例如，connection_to_emr_kafka。
   專有網(wǎng)絡	選擇與EMR集群相同的專有網(wǎng)絡。 如果當前沒有可選擇的專有網(wǎng)絡，請單擊創(chuàng)建專有網(wǎng)絡，前往專有網(wǎng)絡控制臺創(chuàng)建，詳情請參見創(chuàng)建和管理專有網(wǎng)絡。
   交換機	選擇與EMR集群部署在同一專有網(wǎng)絡下的相同交換機。 如果當前可用區(qū)沒有交換機，請單擊虛擬交換機，前往專有網(wǎng)絡控制臺創(chuàng)建，詳情請參見創(chuàng)建和管理交換機。

   當狀態(tài)顯示為已成功時，表示新增網(wǎng)絡連接成功。

步驟三：為EMR集群添加安全組規(guī)則

1. 獲取集群節(jié)點交換機的網(wǎng)段。

   您可以在節(jié)點管理頁面，單擊節(jié)點組名稱，查看關聯(lián)的交換機信息，然后登錄專有網(wǎng)絡管理控制臺，在交換機頁面獲取交換機的網(wǎng)段。

   

2. 添加安全組規(guī)則。

   1. 在集群管理頁面，單擊目標集群的集群ID。

   2. 在基礎信息頁面，單擊集群安全組后面的鏈接。

   3. 在安全組規(guī)則頁面，單擊手動添加，填寫端口范圍和授權對象，然后單擊保存。

      參數(shù)	說明
      端口范圍	填寫9092端口。
      授權對象	填寫前一步驟中獲取的指定交換機的網(wǎng)段。 重要 為防止被外部的用戶攻擊導致安全問題，授權對象禁止填寫為0.0.0.0/0。

步驟四：上傳JAR包至OSS

解壓文件kafka.zip，并將文件中的所有JAR包上傳至OSS，上傳操作請參見簡單上傳。

步驟五：上傳資源文件

1. 在EMR Serverless Spark頁面，單擊左側導航欄中的文件管理。

2. 在文件管理頁面，單擊上傳文件。

3. 在上傳文件對話框中，單擊待上傳文件區(qū)域選擇pyspark_ss_demo.py文件。

步驟六：新建并啟動流任務

1. 在EMR Serverless Spark頁面，單擊左側的數(shù)據(jù)開發(fā)。

2. 單擊新建。

3. 輸入名稱，任務類型選擇Application（流任務） > PySpark，然后單擊確定。

4. 在新建的開發(fā)中，配置以下信息，其余參數(shù)無需配置，然后單擊保存。

   參數(shù)	說明
   主Python資源	選擇前一個步驟中在資源上傳頁面上傳的pyspark_ss_demo.py文件。
   引擎版本	Spark的版本，詳情請參見引擎版本介紹。
   運行參數(shù)	EMR集群core-1-1節(jié)點的內網(wǎng)IP地址。您可以在EMR集群的節(jié)點管理頁面的Core節(jié)點組下查看。
   Spark配置	Spark的配置信息。本文示例如下。 spark.jars oss://path/to/commons-pool2-2.11.1.jar,oss://path/to/kafka-clients-2.8.1.jar,oss://path/to/spark-sql-kafka-0-10_2.12-3.3.1.jar,oss://path/to/spark-token-provider-kafka-0-10_2.12-3.3.1.jar spark.emr.serverless.network.service.name connection_to_emr_kafka 說明 spark.jars：指定Spark任務運行時需要加載的外部JAR包路徑。請根據(jù)實際情況替換為步驟四中上傳的所有JAR包文件路徑。 spark.emr.serverless.network.service.name：指定網(wǎng)絡連接的名稱。請根據(jù)實際情況替換為步驟二中創(chuàng)建的網(wǎng)絡連接名稱。

5. 單擊發(fā)布。

6. 在發(fā)布任務對話框中，單擊確定。

7. 啟動流任務。

   1. 單擊前往運維。

   2. 單擊啟動。

步驟七：查看日志

1. 單擊日志探查頁簽。

2. 在Driver日志列表中，單擊stdOut.log。

   在打開的日志文件中，您可以看到應用程序執(zhí)行的相關信息以及返回的結果。