文档首页/ MapReduce服务 MRS/ 开发指南（LTS版）/ Flink开发指南（普通模式）/ 开发Flink应用/ PyFlink样例程序/ 使用Python提交Flink普通作业
更新时间：2024-08-03 GMT+08:00
查看PDF
分享

使用Python提交Flink普通作业

  1. 获取样例工程“flink-examples/pyflink-example/pyflink-kafka”中的“pyflink-kafka.py”和“insertData2kafka.sql”。
  2. 参考准备本地应用开发环境将准备好的Python虚拟环境打包，获取“venv.zip”文件。 

    zip -q -r venv.zip venv/

  3. 以root用户登录主管理节点，将1和2获取的“venv.zip”、“pyflink-kafka.py”和“insertData2kafka.sql”文件上传至客户端环境。

    • per-job模式：将上述文件上传到“客户端安装目录/Flink/flink”。
    • yarn-application模式：将上述文件和“flink-connector-kafka-实际版本号.jar”包上传到“客户端安装目录/Flink/flink/yarnship”。

  4. 修改“pyflink-kafka.py”中的“specific_jars”路径。

    • per-job模式：修改为SQL文件的实际路径。如：file:///客户端安装目录/Flink/flink/lib/flink-connector-kafka-实际版本号.jar
    • yarn-application模式：修改为：file://"+os.getcwd()+"/../../../../yarnship/flink-connector-kafka-实际版本号.jar

  5. 修改“pyflink-kafka.py”中的“file_path”路径。

    • per-job模式：修改为sql文件的实际路径。如：客户端安装目录/Flink/flink/insertData2kafka.sql
    • yarn-application模式：修改为os.getcwd() + "/../../../../yarnship/insertData2kafka.sql"

  6. 执行以下命令指定运行环境。 

    export PYFLINK_CLIENT_EXECUTABLE=venv.zip/venv/bin/python3

  7. 执行以下命令运行程序。

    • per-job模式： 
      ./bin/flink run --detached -t yarn-per-job -Dyarn.application.name=py_kafka -pyarch venv.zip -pyexec venv.zip/venv/bin/python3 -py pyflink-kafka.py

      运行结果：

    • yarn-application模式 
      ./bin/flink run-application --detached -t yarn-application -Dyarn.application.name=py_kafka -Dyarn.ship-files=/opt/client/Flink/flink/yarnship/ -pyarch yarnship/venv.zip -pyexec venv.zip/venv/bin/python3 -pyclientexec venv.zip/venv/bin/python3 -pyfs yarnship -pym pyflink-kafka

      运行结果：

父主题： PyFlink样例程序