?????? pyflink1.11.0window

?????????????? Wed, 15 Jul 2020 02:31:20 -0700

&nbsp; &nbsp; &nbsp;??????????




------------------&nbsp;????????&nbsp;------------------
??????:                                                                         
                                               "user-zh"                        
                                                            
<acqua....@gmail.com&gt;;
????????:&nbsp;2020??7??15??(??????) ????5:23
??????:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;;

????:&nbsp;Re: pyflink1.11.0window



??????????????kafka????json???????????? ????????????????????es?? 
???????????????????? ?????????????????? ??????????????????????????????

from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
from pyflink.table import StreamTableEnvironment, DataTypes, EnvironmentSettings
from pyflink.table.udf import udf


@udf(input_types=[DataTypes.INT()], result_type=DataTypes.STRING())
def platform_code_to_name(code):
&nbsp;&nbsp;&nbsp; return "mobile" if code == 0 else "pc"


def log_processing():
&nbsp;&nbsp;&nbsp; env = StreamExecutionEnvironment.get_execution_environment()
&nbsp;&nbsp;&nbsp; 
env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
&nbsp;&nbsp;&nbsp; env_settings = 
EnvironmentSettings.Builder().use_blink_planner().build()
&nbsp;&nbsp;&nbsp; t_env = 
StreamTableEnvironment.create(stream_execution_environment=env,
environment_settings=env_settings)

&nbsp;&nbsp;&nbsp; source_ddl = """
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; CREATE TABLE 
payment_msg(
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 createTime VARCHAR,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 rt as TO_TIMESTAMP(createTime),
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 orderId BIGINT,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 payAmount DOUBLE,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 payPlatform INT,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 paySource INT,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 WATERMARK FOR rt as rt - INTERVAL '2' SECOND
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ) WITH (
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
'connector.type' = 'kafka',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
'connector.version' = 'universal',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
'connector.topic' = 'payment_msg_2',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
'connector.properties.bootstrap.servers' = '0.0.0.0:9092',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
'connector.properties.group.id' = 'test_3',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
'connector.startup-mode' = 'latest-offset',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
'format.type' = 'json'
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; )
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; """
&nbsp;&nbsp;&nbsp; t_env.sql_update(source_ddl)

&nbsp;&nbsp;&nbsp; es_sink_ddl = """
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; CREATE TABLE 
es_sink (
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; platform 
VARCHAR,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; pay_amount 
DOUBLE,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; rowtime 
TIMESTAMP(3)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ) with (
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.type' = 'elasticsearch',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.version' = '7',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.hosts' = 'http://localhost:9200',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.index' = 'platform_pay_amount_1',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.document-type' = 'payment',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'update-mode' = 'upsert',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.flush-on-checkpoint' = 'true',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.key-delimiter' = '$',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.key-null-literal' = 'n/a',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.bulk-flush.max-size' = '42mb',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.bulk-flush.max-actions' = '32',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.bulk-flush.interval' = '1000',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'connector.bulk-flush.backoff.delay' = '1000',
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 'format.type' = 'json'
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; )
&nbsp;&nbsp;&nbsp; """

&nbsp;&nbsp;&nbsp; t_env.sql_update(es_sink_ddl)

&nbsp;&nbsp;&nbsp; t_env.register_function('platformcodetoname', 
platform_code_to_name)

&nbsp;&nbsp;&nbsp; query = """
&nbsp;&nbsp;&nbsp; select platformcodetoname(payPlatform) as platform, 
sum(payAmount)
as pay_amount, cast(tumble_start(rt, interval '5' seconds) as BIGINT)
as rowtime
&nbsp;&nbsp;&nbsp; from payment_msg
&nbsp;&nbsp;&nbsp; group by tumble(rt, interval '5' seconds), payPlatform
&nbsp;&nbsp;&nbsp; """

&nbsp;&nbsp;&nbsp; count_result = t_env.sql_query(query)

&nbsp;&nbsp;&nbsp; t_env.create_temporary_view('windowed_values', count_result)

&nbsp;&nbsp;&nbsp; query2 = """
&nbsp;&nbsp;&nbsp; select platform, last_value(pay_amount), rowtime from
windowed_values group by platform, rowtime
&nbsp;&nbsp;&nbsp; """

&nbsp;&nbsp;&nbsp; final_result = t_env.sql_query(query2)

&nbsp;&nbsp;&nbsp; final_result.execute_insert(table_path='es_sink')


if __name__ == '__main__':
&nbsp;&nbsp;&nbsp; log_processing()


?????????????? <1129656...@qq.com&gt; ??2020??7??15?????? ????4:40??????

&gt; &amp;nbsp;Shuiqiang????????
&gt; &amp;nbsp; &amp;nbsp;
&gt; 
&amp;nbsp;hi????????????????????????????????????????????????????????2-from_kafka_to_kafka.py??????????????????????????????????????????????demo????????????????????????????????????demo??????????????????????????????????????????????????????????????????
&gt;
&gt;
&gt; ????????????????????????????
&gt;
&gt;
&gt; ??????
&gt;
&gt;
&gt;
&gt;
&gt; ------------------&amp;nbsp;????????&amp;nbsp;------------------
&gt; ??????:
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 "user-zh"
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 <
&gt; acqua....@gmail.com&amp;gt;;
&gt; ????????:&amp;nbsp;2020??7??15??(??????) ????11:25
&gt; ??????:&amp;nbsp;"user-zh"<user-zh@flink.apache.org&amp;gt;;
&gt;
&gt; ????:&amp;nbsp;Re: pyflink1.11.0window
&gt;
&gt;
&gt;
&gt; ????sql????
&gt; select platformcodetoname(payPlatform) as platform, sum(payAmount) as
&gt; pay_amount, cast(tumble_start(rt, interval '5' seconds) as BIGINT) as
&gt; rowtime
&gt; from payment_msg group by tumble(rt, interval '5' seconds), payPlatform
&gt; ????query ????5s??tumble????????????
&gt;
&gt; ?????????????? <1129656...@qq.com&amp;gt; ??2020??7??15?????? 
????11:10??????
&gt;
&gt; &amp;gt; Shuiqiang????????
&gt; &amp;gt; &amp;amp;nbsp;
&gt; 
&amp;amp;nbsp;??????????????????????????????????????????????????????????????????????????????????????window??
&gt; &amp;gt;
&gt; &amp;gt;
&gt; &amp;gt; 
------------------&amp;amp;nbsp;????????&amp;amp;nbsp;------------------
&gt; &amp;gt; ??????:
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
&gt; "user-zh"
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
&gt; <
&gt; &amp;gt; acqua....@gmail.com&amp;amp;gt;;
&gt; &amp;gt; ????????:&amp;amp;nbsp;2020??7??15??(??????) ????10:51
&gt; &amp;gt; 
??????:&amp;amp;nbsp;"user-zh"<user-zh@flink.apache.org&amp;amp;gt;;
&gt; &amp;gt;
&gt; &amp;gt; ????:&amp;amp;nbsp;Re: pyflink1.11.0window
&gt; &amp;gt;
&gt; &amp;gt;
&gt; &amp;gt;
&gt; &amp;gt; ??????????
&gt; &amp;gt; ??????????org.apache.flink.table.api.ValidationException: A 
tumble window
&gt; &amp;gt; expects a size value literal.
&gt; &amp;gt; ????????????tumble window????????????????????
&gt; &amp;gt;
&gt; &amp;gt; Best,
&gt; &amp;gt; Shuiqiang
&gt; &amp;gt;
&gt; &amp;gt; ?????????????? <1129656...@qq.com&amp;amp;gt; ??2020??7??15?????? 
????10:27??????
&gt; &amp;gt;
&gt; &amp;gt; &amp;amp;gt; ??????
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
&amp;amp;amp;nbsp;??????????????????????source??????????????????????????????????????????????????window??????????????????????????????
&gt; &amp;gt; &amp;amp;gt; Traceback (most recent call last):
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; File "tou.py", line 71, in 
<module&amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
from_kafka_to_kafka_demo()
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; File "tou.py", line 21, in
&gt; from_kafka_to_kafka_demo
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; .select(" 
id,&amp;amp;amp;nbsp;
&gt; time1 , time1 ")\
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; File
&gt; &amp;gt; &amp;amp;gt;
&gt; "/usr/local/lib/python3.7/site-packages/pyflink/table/table.py", line
&gt; &amp;gt; 907,
&gt; &amp;gt; &amp;amp;gt; in select
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; return
&gt; Table(self._j_table.select(fields),
&gt; &amp;gt; self._t_env)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; File
&gt; &amp;gt; "/usr/local/lib/python3.7/site-packages/py4j/java_gateway.py",
&gt; &amp;gt; &amp;amp;gt; line 1286, in __call__
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; answer, 
self.gateway_client,
&gt; self.target_id,
&gt; &amp;gt; self.name)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; File
&gt; &amp;gt; &amp;amp;gt;
&gt; "/usr/local/lib/python3.7/site-packages/pyflink/util/exceptions.py",
&gt; &amp;gt; line
&gt; &amp;gt; &amp;amp;gt; 147, in deco
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; return f(*a, 
**kw)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; File
&gt; &amp;gt; "/usr/local/lib/python3.7/site-packages/py4j/protocol.py",
&gt; &amp;gt; &amp;amp;gt; line 328, in get_return_value
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
format(target_id, ".", name),
&gt; value)
&gt; &amp;gt; &amp;amp;gt; py4j.protocol.Py4JJavaError: An error occurred while 
calling
&gt; &amp;gt; o26.select.
&gt; &amp;gt; &amp;amp;gt; : org.apache.flink.table.api.ValidationException: A 
tumble
&gt; window
&gt; &amp;gt; expects
&gt; &amp;gt; &amp;amp;gt; a size value literal.
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
org.apache.flink.table.operations.utils.AggregateOperationFactory.getAsValueLiteral(AggregateOperationFactory.java:384)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
org.apache.flink.table.operations.utils.AggregateOperationFactory.validateAndCreateTumbleWindow(AggregateOperationFactory.java:302)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
org.apache.flink.table.operations.utils.AggregateOperationFactory.createResolvedWindow(AggregateOperationFactory.java:236)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
org.apache.flink.table.operations.utils.OperationTreeBuilder.windowAggregate(OperationTreeBuilder.java:250)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
org.apache.flink.table.api.internal.TableImpl$WindowGroupedTableImpl.select(TableImpl.java:794)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
org.apache.flink.table.api.internal.TableImpl$WindowGroupedTableImpl.select(TableImpl.java:781)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt; sun.reflect.NativeMethodAccessorImpl.invoke0(Native 
Method)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt; java.lang.reflect.Method.invoke(Method.java:498)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; at
&gt; &amp;gt; java.lang.Thread.run(Thread.java:748)
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; def register_rides_source(st_env):
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; source_ddl = \
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; """
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; create table 
source1(
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp;id int,
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp;time1 timestamp,
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp;type string,
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp;WATERMARK FOR
&gt; time1 as time1 -
&gt; &amp;gt; INTERVAL '2' SECOND
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp;) with (
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
'connector.type' = 'kafka',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 'update-mode' 
= 'append',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
'connector.topic' = 'tp1',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; 'connector.properties.bootstrap.servers' =
&gt; &amp;gt; 'localhost:9092',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; 'connector.properties.zookeeper.connect' =
&gt; &amp;gt; 'localhost:2181',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 'format.type' 
= 'json',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
'format.derive-schema' =
&gt; 'true',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
'connector.version' =
&gt; 'universal'
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp;)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; """
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
st_env.sql_update(source_ddl)
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; s_env =
&gt; &amp;gt; &amp;amp;gt; 
StreamExecutionEnvironment.get_execution_environment()
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
s_env.set_parallelism(1)
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; st_env =
&gt; StreamTableEnvironment.create(s_env)
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
register_rides_source(st_env)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
register_rides_sink(st_env)
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
st_env.from_path("source1")\
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;gt; 
.window(Tumble.over("2.secends").on("time1").alias("w")) \
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; .group_by("w") \
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; .select(" id,&amp;amp;amp;nbsp;
&gt; &amp;gt; time1 , time1 ")\
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; .insert_into("sink1")
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; 
&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;nbsp; &amp;amp;amp;nbsp;
&gt; st_env.execute("2-from_kafka_to_kafka")
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; ????????
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; 
------------------&amp;amp;amp;nbsp;????????&amp;amp;amp;nbsp;------------------
&gt; &amp;gt; &amp;amp;gt; ??????:
&gt; &amp;gt;
&gt; 
&amp;amp;gt;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;
&gt; &amp;gt; "user-zh"
&gt; &amp;gt;
&gt; 
&amp;amp;gt;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;
&gt; &amp;gt; <
&gt; &amp;gt; &amp;amp;gt; acqua....@gmail.com&amp;amp;amp;gt;;
&gt; &amp;gt; &amp;amp;gt; ????????:&amp;amp;amp;nbsp;2020??7??10??(??????) 
????9:17
&gt; &amp;gt; &amp;amp;gt; 
??????:&amp;amp;amp;nbsp;"user-zh"<user-zh@flink.apache.org
&gt; &amp;amp;amp;gt;;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; ????:&amp;amp;amp;nbsp;Re: pyflink1.11.0window
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; ??????????
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; ????source ddl????????time1?? time attribute????
&gt; &amp;gt; &amp;amp;gt; create table source1(
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
&gt; id
&gt; &amp;gt; int,
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
&gt; &amp;gt; time1 timestamp,
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
&gt; type
&gt; &amp;gt; string,
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
&gt; &amp;gt; WATERMARK FOR time1 as time1 -
&gt; &amp;gt; &amp;amp;gt; INTERVAL '2' SECOND
&gt; &amp;gt; &amp;amp;gt; ) with (...)
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; ?????????????? <1129656...@qq.com&amp;amp;amp;gt; 
??2020??7??10??????
&gt; ????8:43??????
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt;
&gt; 
------------------&amp;amp;amp;amp;nbsp;????????&amp;amp;amp;amp;nbsp;------------------
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; ??????:
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
&amp;amp;amp;gt;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;gt; "??????????????"
&gt; &amp;gt; &amp;amp;gt;
&gt; &amp;gt;
&gt; 
&amp;amp;amp;gt;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;gt; <
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; 
1129656...@qq.com&amp;amp;amp;amp;gt;;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; 
????????:&amp;amp;amp;amp;nbsp;2020??7??9??(??????) ????5:08
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; 
??????:&amp;amp;amp;amp;nbsp;"godfrey he"<
&gt; godfre...@gmail.com
&gt; &amp;gt; &amp;amp;amp;amp;gt;;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; 
????:&amp;amp;amp;amp;nbsp;pyflink1.11.0window
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; ??????
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp;????????pyflink1.11????????window????????????
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; :
&gt; org.apache.flink.table.api.ValidationException: A group
&gt; &amp;gt; window
&gt; &amp;gt; &amp;amp;gt; expects a
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; time attribute for grouping in a 
stream
&gt; environment.
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; 
????????????????????????????????????????????????????????????????????????????????????????
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; ????????
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; ????
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; def from_kafka_to_kafka_demo():
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp; s_env =
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; StreamExecutionEnvironment.get_execution_environment()
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; s_env.set_parallelism(1)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp; # use
&gt; blink table planner
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp; st_env =
&gt; &amp;gt; StreamTableEnvironment.create(s_env)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp; #
&gt; register source and sink
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; register_rides_source(st_env)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; register_rides_sink(st_env)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; st_env.from_path("source1")\
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp; &amp;amp;amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; .window(Tumble.over("1.secends").on("time1").alias("w")) \
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp; &amp;amp;amp;amp;nbsp;
&gt; &amp;gt; .group_by("w") \
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp; &amp;amp;amp;amp;nbsp;
&gt; &amp;gt; .select(" id,&amp;amp;amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;gt; time1 , time1 ")\
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp; &amp;amp;amp;amp;nbsp;
&gt; &amp;gt; .insert_into("sink1")
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;gt; st_env.execute("2-from_kafka_to_kafka")
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; def register_rides_source(st_env):
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; source_ddl = \
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp; '''
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp; create
&gt; table source1(
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp; &amp;amp;amp;amp;nbsp;
&gt; &amp;gt; id int,
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp;time1 timestamp,
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp;type string
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp;) with (
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; 'connector.type' = 'kafka',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; 'update-mode' = 'append',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; 'connector.topic' = 'tp1',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;gt; 'connector.properties.bootstrap.servers' =
&gt; &amp;gt; &amp;amp;gt; 'localhost:9092'
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp;)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp; '''
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; st_env.sql_update(source_ddl)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; def register_rides_sink(st_env):
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp; sink_ddl
&gt; = \
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp; '''
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp; create
&gt; table sink1(
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp; &amp;amp;amp;amp;nbsp;
&gt; &amp;gt; id int,
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp;time1 timestamp,
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp;time2 timestamp
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp;) with (
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; 'connector.type' = 'kafka',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; 'update-mode' = 'append',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; 'connector.topic' = 'tp3',
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;gt; 'connector.properties.bootstrap.servers' =
&gt; &amp;gt; &amp;amp;gt; 'localhost:9092'
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; &amp;amp;amp;amp;nbsp;)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp; '''
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; st_env.sql_update(sink_ddl)
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; if __name__ == '__main__':
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;
&gt; from_kafka_to_kafka_demo()
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt;
&gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; 
&amp;amp;amp;amp;nbsp;

?????? pyflink1.11.0window

回复