原始代碼長這樣:

t = t_env.from_elements([(1, 2, 'Welcome'), (2, 3, 'To'), (3, 4, 'PyFlink')], ['a', 'b', 'c'])t_env.connect(FileSystem().path(sink_path))\.with_format(OldCsv().field_delimiter(',').field("add1", DataTypes.BIGINT()).field("add2", DataTypes.BIGINT()).field("add3", DataTypes.BIGINT()).field("add4", DataTypes.BIGINT()).field("b", DataTypes.BIGINT()).field("c", DataTypes.STRING()))\.with_schema(Schema().field("add1", DataTypes.BIGINT()).field("add2", DataTypes.BIGINT()).field("add3", DataTypes.BIGINT()).field("add4", DataTypes.BIGINT()).field("b", DataTypes.BIGINT()).field("c", DataTypes.STRING()))\.register_table_sink("pyflink_sink")

看上去人畜無害，

報錯:

AttributeError: 'StreamTableDescriptor' object has no attribute 'register_table_sink'

原因:

根據[1]中所述,flink1.11中已經刪除了

解決方案:

.register_table_sink("pyflink_sink")

改成

.create_temporary_table("pyflink_sink")

?

[1]Kafka connector with PyFlink

總結

以上是生活随笔為你收集整理的‘StreamTableDescriptor‘ object has no attribute ‘register_table_sink‘的全部內容，希望文章能夠幫你解決所遇到的問題。

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