# 使用Python提交Flink普通作业
1. 获取样例工程"flink-examples/pyflink-example/pyflink-kafka"中的"pyflink-kafka.py"和"insertData2kafka.sql"。
2. 参考[准备本地应用开发环境](https://support.huaweicloud.com/devg-lts-mrs/mrs_07_050006.html)将准备好的Python虚拟环境打包，获取"venv.zip"文件。
   
   ```
   zip -q -r venv.zip venv/
   ```
   
   
3. 以**root** 用户登录主管理节点，将[1]和[2]获取的"venv.zip"、"pyflink-kafka.py"和"insertData2kafka.sql"文件上传至客户端环境。
   
   yarn-application模式：将上述文件和"flink-connector-kafka-*实际版本号* .jar"包上传到"*客户端安装目录*/Flink/flink/yarnship"。
   
   
4. 修改"pyflink-kafka.py"中的"specific_jars"路径。 
   yarn-application模式：修改为：file://"+os.getcwd()+"/../../../../yarnship/flink-connector-kafka-*实际版本号*.jar
   
   
5. 修改"pyflink-kafka.py"中的"file_path"路径。 
   yarn-application模式：修改为os.getcwd() + "/../../../../yarnship/insertData2kafka.sql"
   
   
6. 执行以下命令指定运行环境。 
   ```
   export PYFLINK_CLIENT_EXECUTABLE=venv.zip/venv/bin/python3
   ```
   
   
7. 执行以下命令运行程序。 
   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
   ```
   运行结果：
   ![](https://support.huaweicloud.com/devg-lts-mrs/zh-cn_image_0000001695475114.png "点击放大")