?實驗記錄:

①啟動hadoop

②$FLINK_HOME/bin/pyflink-shell.sh yarn

?

?使用了Table API的PyFlink完整代碼如下:

import tempfile import os import shutil sink_path = tempfile.gettempdir() + '/streaming.csv' if os.path.exists(sink_path):if os.path.isfile(sink_path):os.remove(sink_path)else:shutil.rmtree(sink_path)s_env.set_parallelism(1) t = st_env.from_elements([(1, 'hi3', 'hello3'), (2, 'hi3', 'hello3')], ['a', 'b', 'c']) st_env.connect(FileSystem().path(sink_path)).with_format(OldCsv().field_delimiter(',').field("a", DataTypes.BIGINT()).field("b", DataTypes.STRING()).field("c", DataTypes.STRING())).with_schema(Schema().field("a", DataTypes.BIGINT()).field("b", DataTypes.STRING()).field("c", DataTypes.STRING())).register_table_sink("stream_sink")t.select("a + 1, b, c").insert_into("stream_sink")st_env.execute("stream_job")

實驗結果:

在集群中的某個節點(我的是Laptop節點)中的/tmp目錄下

(Python3.6) appleyuchi@Laptop:tmp$ cat streaming.csv ??
2,hi3,hello3
3,hi3,hello3

?

怎么看是不是per-job模式呢？如下圖右側的yarn界面

?

總結

以上是生活随笔為你收集整理的pyflink shell on yarn在per-job cluster模式下的实验记录的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。