官方文档
首页 开源大数据平台E-MapReduce EMR on ECS 实践教程 数据迁移和同步 通过PyFlink作业处理Kafka数据

通过PyFlink作业处理Kafka数据

更新时间：
复制 MD 格式
产品详情

本文介绍在阿里云E-MapReduce创建的包含Flink和kafka服务的DataFlow集群中，如何通过PyFlink来处理Kafka中的实时流数据。

前提条件

  • 已完成云账号的授权，详情请参见角色授权。

  • 已创建包含Flink和Kafka服务的DataFlow集群，创建详情请参见创建集群。

    本文以EMR-5.17.3版本为例。

步骤一：创建Topic

本示例将创建两个名称为payment-msg和result的Topic。

  1. 登录集群的Master节点，详情请参见登录集群。

  2. 执行如下命令，创建Topic。

    kafka-topics.sh --partitions 3 --replication-factor 2 --bootstrap-server core-1-1:9092 --topic payment-msg --create
kafka-topics.sh --partitions 3 --replication-factor 2 --bootstrap-server core-1-1:9092 --topic result --create
    说明

    创建Topic后，请保留该登录窗口，后续步骤仍将使用。

步骤二：准备测试数据

执行如下命令，不断生成测试数据。

python3 -m pip install kafka
rm -rf produce_data.py
cat>produce_data.py<<EOF
import random
import time, calendar
from random import randint
from kafka import KafkaProducer
from json import dumps
from time import sleep
def write_data():
    data_cnt = 20000
    order_id = calendar.timegm(time.gmtime())
    max_price = 100000
    topic = "payment-msg"
    producer = KafkaProducer(bootstrap_servers=['core-1-1:9092'],
                             value_serializer=lambda x: dumps(x).encode('utf-8'))
    for i in range(data_cnt):
        ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        rd = random.random()
        order_id += 1
        pay_amount = max_price * rd
        pay_platform = 0 if random.random() < 0.9 else 1
        province_id = randint(0, 6)
        cur_data = {"createTime": ts, "orderId": order_id, "payAmount": pay_amount, "payPlatform": pay_platform, "provinceId": province_id}
        producer.send(topic, value=cur_data)
        sleep(0.5)
if __name__ == '__main__':
    write_data()
EOF
python3 produce_data.py

步骤三：创建并运行PyFlink作业

  1. 登录集群的Master节点，详情请参见登录集群。

  2. 执行如下命令，生成PyFlink作业文件。

    rm -rf job.py
cat>job.py<<EOF
import os
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
from pyflink.table import StreamTableEnvironment, DataTypes
from pyflink.table.udf import udf
provinces = ("beijing", "shanghai", "hangzhou", "shenzhen", "jiangxi", "chongqing", "xizang")
@udf(input_types=[DataTypes.INT()], result_type=DataTypes.STRING())
def province_id_to_name(id):
    return provinces[id]
#请根据创建的DataFlow集群，输入以下信息。
def log_processing():
    kafka_servers = "core-1-1:9092"
    source_topic = "payment-msg"
    sink_topic = "result"
    kafka_consumer_group_id = "test"
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
    t_env = StreamTableEnvironment.create(stream_execution_environment=env)
    t_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True)
    source_ddl = f"""
            CREATE TABLE payment_msg(
                createTime VARCHAR,
                rt as TO_TIMESTAMP(createTime),
                orderId BIGINT,
                payAmount DOUBLE,
                payPlatform INT,
                provinceId INT,
                WATERMARK FOR rt as rt - INTERVAL '2' SECOND
            ) WITH (
              'connector' = 'kafka',
              'topic' = '{source_topic}',
              'properties.bootstrap.servers' = '{kafka_servers}',
              'properties.group.id' = '{kafka_consumer_group_id}',
              'scan.startup.mode' = 'latest-offset',
              'format' = 'json'
            )
            """
    kafka_sink_ddl = f"""
            CREATE TABLE kafka_sink (
            province VARCHAR,
            pay_amount DOUBLE,
            rowtime TIMESTAMP(3)
            ) with (
              'connector' = 'kafka',
              'topic' = '{sink_topic}',
              'properties.bootstrap.servers' = '{kafka_servers}',
              'properties.group.id' = '{kafka_consumer_group_id}',
              'scan.startup.mode' = 'latest-offset',
              'format' = 'json'
            )
    """
    t_env.execute_sql(source_ddl)
    t_env.execute_sql(kafka_sink_ddl)
    t_env.register_function('province_id_to_name', province_id_to_name)
    sink_ddl = """
    insert into kafka_sink select province_id_to_name(provinceId) as province, sum(payAmount) as pay_amount, tumble_start(rt, interval '5' second) as rowtime
    from payment_msg
    group by tumble(rt, interval '5' second), provinceId
    """
    t_env.execute_sql(sink_ddl)
if __name__ == '__main__':
    log_processing()
EOF

    请您根据集群的实际情况，修改如下参数。

    参数

    描述

    kafka_servers

    指定DataFlow集群中Broker节点的内网IP地址和端口号，端口号默认为9092。

    source_topic

    源表的Kafka Topic，本文示例为payment-msg。

    sink_topic

    结果表的Kafka Topic，本文示例为result。

  3. 执行以下命令，运行PyFlink作业。

    flink run -t yarn-per-job -py job.py -j /opt/apps/FLINK/flink-current/opt/connectors/kafka/flink-sql-connector-kafka-*.jar

步骤四：查看作业信息

  1. 访问YARN UI，访问Web UI详情请参见通过控制台访问开源组件Web界面。

    通过访问YARN UI查看Flink作业的信息。

  2. 在Hadoop控制台，单击作业的ID。

    您可以查看作业运行详情。进入 All Applications 页面，在应用列表的 ID 列中找到目标 Flink 作业对应的 ID。

    详细信息如下。在 YARN ResourceManager 的 Application Overview 页面中，可查看作业的 User、Name（Flink per-job cluster）、Application Type（Apache Flink）、YarnApplicationState（RUNNING）、Queue（default）等运行信息。单击 Tracking URL 对应的 ApplicationMaster 链接可跳转至 Flink Web UI 查看作业详情。

  3. 可选：单击Tracking URL后面的链接，进入Apache Flink Dashboard页面。

    您可以查看详情的作业信息。

步骤五：查看输出数据

  1. 登录集群的Master节点。详情请参见登录集群。

  2. 执行如下命令，查看result的数据。

    kafka-console-consumer.sh --bootstrap-server core-1-1:9092 --topic result

    返回信息如下。命令执行后，返回如下类似结果。

    {"province":"shanghai","pay_amount":75736.6129370...,"rowtime":"2024-01-11 14:35:00"}
{"province":"beijing","pay_amount":134007.5956410...,"rowtime":"2024-01-11 14:35:00"}
{"province":"chongqing","pay_amount":7589.3463450...,"rowtime":"2024-01-11 14:35:00"}
{"province":"chongqing","pay_amount":71427.570820...,"rowtime":"2024-01-11 14:35:00"}
{"province":"beijing","pay_amount":109241.1396200...,"rowtime":"2024-01-11 14:35:05"}
{"province":"shanghai","pay_amount":88272.9180790...,"rowtime":"2024-01-11 14:35:05"}
{"province":"jiangxi","pay_amount":37021.35826730...,"rowtime":"2024-01-11 14:35:05"}
{"province":"hangzhou","pay_amount":188936.694150...,"rowtime":"2024-01-11 14:35:05"}
{"province":"shenzhen","pay_amount":31858.9857300...,"rowtime":"2024-01-11 14:35:05"}
{"province":"shanghai","pay_amount":56345.8958370...,"rowtime":"2024-01-11 14:35:10"}
{"province":"beijing","pay_amount":89964.49844080...,"rowtime":"2024-01-11 14:35:10"}
{"province":"jiangxi","pay_amount":96612.80666800...,"rowtime":"2024-01-11 14:35:10"}
{"province":"chongqing","pay_amount":188225.17380...,"rowtime":"2024-01-11 14:35:10"}
{"province":"xizang","pay_amount":132150.87189850...,"rowtime":"2024-01-11 14:35:15"}
{"province":"beijing","pay_amount":103155.9561520...,"rowtime":"2024-01-11 14:35:15"}
{"province":"hangzhou","pay_amount":99597.3313220...,"rowtime":"2024-01-11 14:35:15"}
{"province":"chongqing","pay_amount":171639.14540...,"rowtime":"2024-01-11 14:35:15"}
{"province":"jiangxi","pay_amount":13950.96936230...,"rowtime":"2024-01-11 14:35:15"}
{"province":"hangzhou","pay_amount":94148.8945930...,"rowtime":"2024-01-11 14:35:20"}
{"province":"jiangxi","pay_amount":59460.30738120...,"rowtime":"2024-01-11 14:35:20"}
{"province":"shanghai","pay_amount":33501.8557620...,"rowtime":"2024-01-11 14:35:20"}
{"province":"xizang","pay_amount":86399.980186180...,"rowtime":"2024-01-11 14:35:20"}
{"province":"beijing","pay_amount":241098.9064580...,"rowtime":"2024-01-11 14:35:20"}
{"province":"chongqing","pay_amount":86470.525200...,"rowtime":"2024-01-11 14:35:20"}

相关文档

如需全面了解Kafka topic中数据的生产和消费机制，详情请参见Apache Kafka SQL连接器。

上一篇：通过Spark Streaming作业处理Kafka数据下一篇：使用JindoDistCp工具迁移第三方云对象存储数据到OSS/OSS-HDFS