通过Serverless Spark提交PySpark流任务

在大数据快速发展的时代,流式处理技术对于实时数据分析至关重要。EMR Serverless Spark提供了一个强大而可扩展的平台,它不仅简化了实时数据处理流程,还免去了服务器管理的烦恼,提升了效率。本文将指导您使用EMR Serverless Spark提交PySpark流式任务,展示其在流处理方面的易用性和可运维性。

前提条件

已创建工作空间,详情请参见创建工作空间

操作流程

步骤一:创建实时数据流集群并产生消息

  1. 在EMR on ECS页面,创建包含Kafka服务的实时数据流集群,详情请参见创建集群

  2. 登录EMR集群的Master节点,详情请参见登录集群

  3. 执行以下命令,切换目录。

    cd /var/log/emr/taihao_exporter
  4. 执行以下命令,创建Topic。

    # 创建名为taihaometrics的Topic,分区数10,副本因子2。
    kafka-topics.sh --partitions 10 --replication-factor 2 --bootstrap-server core-1-1:9092 --topic taihaometrics --create
  5. 执行以下命令,发送消息。

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

步骤二:新增网络连接

  1. 进入网络连接页面。

    1. 在EMR控制台的左侧导航栏,选择EMR Serverless > Spark

    2. Spark页面,单击目标工作空间名称。

    3. EMR Serverless Spark页面,单击左侧导航栏中的网络连接

  2. 网络连接页面,单击新增网络连接

  3. 新增网络连接对话框中,配置以下信息,单击确定

    参数

    说明

    连接名称

    输入新增连接的名称。例如,connection_to_emr_kafka。

    专有网络

    选择与EMR集群相同的专有网络。

    如果当前没有可选择的专有网络,请单击创建专有网络,前往专有网络控制台创建,详情请参见创建和管理专有网络

    交换机

    选择与EMR集群部署在同一专有网络下的相同交换机。

    如果当前可用区没有交换机,请单击虚拟交换机,前往专有网络控制台创建,详情请参见创建和管理交换机

    状态显示为已成功时,表示新增网络连接成功。

步骤三:为EMR集群添加安全组规则

  1. 获取集群节点交换机的网段。

    您可以在节点管理页面,单击节点组名称,查看关联的交换机信息,然后登录专有网络管理控制台,在交换机页面获取交换机的网段。

    image

  2. 添加安全组规则。

    1. 集群管理页面,单击目标集群的集群ID。

    2. 基础信息页面,单击集群安全组后面的链接。

    3. 在安全组规则页面,单击手动添加,填写端口范围和授权对象,然后单击保存

      参数

      说明

      端口范围

      填写9092端口。

      授权对象

      填写前一步骤中获取的指定交换机的网段。

      重要

      为防止被外部的用户攻击导致安全问题,授权对象禁止填写为0.0.0.0/0。

步骤四:上传JAR包至OSS

解压文件kafka.zip,并将文件中的所有JAR包上传至OSS,上传操作请参见简单上传

步骤五:上传资源文件

  1. 在EMR Serverless Spark页面,单击左侧导航栏中的文件管理

  2. 文件管理页面,单击上传文件

  3. 上传文件对话框中,单击待上传文件区域选择pyspark_ss_demo.py文件。

步骤六:新建并启动流任务

  1. 在EMR Serverless Spark页面,单击左侧的数据开发

  2. 单击新建

  3. 输入名称,任务类型选择Application(流任务) > PySpark,然后单击确定

  4. 在新建的开发中,配置以下信息,其余参数无需配置,然后单击保存

    参数

    说明

    主Python资源

    选择前一个步骤中在资源上传页面上传的pyspark_ss_demo.py文件。

    引擎版本

    Spark的版本,详情请参见引擎版本介绍

    运行参数

    EMR集群core-1-1节点的内网IP地址。您可以在EMR集群的节点管理页面的Core节点组下查看。

    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包路径。请根据实际情况替换为步骤四中上传的所有JAR包文件路径。

    • spark.emr.serverless.network.service.name:指定网络连接的名称。请根据实际情况替换为步骤二中创建的网络连接名称。

  5. 单击发布

  6. 发布任务对话框中,单击确定

  7. 启动流任务。

    1. 单击前往运维

    2. 单击启动

步骤七:查看日志

  1. 单击日志探查页签。

  2. Driver日志列表中,单击stdOut.log

    在打开的日志文件中,您可以看到应用程序执行的相关信息以及返回的结果。

    image

相关文档

PySpark的开发流程示例,请参见PySpark开发快速入门