python和flink相關的網上例子比較少

先溫固下shell中這些環境變量吧。

python ?flink shell變量	類型	scala?flink shell變量	類型
s_env	pyflink.datastream.stream_execution_environment.StreamExecutionEnvironment	senv	StreamExecutionEnvironment
st_env	pyflink.table.table_environment.StreamTableEnvironment	stenv	StreamTableEnvironmentImpl
b_env	class 'pyflink.dataset.execution_environment.ExecutionEnvironment	benv	ExecutionEnvironment
bt_env	pyflink.table.table_environment.BatchTableEnvironment	btenv	BatchTableEnvironmentImpl

注:

scala獲取類型舉例:

scala> senv.getClass.getSimpleName
res2: String = StreamExecutionEnvironment

####################################################################################################

啟動命令:

pyflink-shell.sh local

?交互模式中輸入的代碼(Python DataStream API):

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, 'hi', 'hello'), (2, 'hi', 'hello')], ['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")

實驗結果:

/tmp/streaming.csv

打開后如下:

?

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

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

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