I got why regarding the simplified question - the dummy parser should return key(string)-value(string), otherwise it fails the result_type spec
On Fri, Mar 19, 2021 at 3:37 PM Yik San Chan <evan.chanyik...@gmail.com> wrote: > Hi Dian, > > I simplify the question in > https://stackoverflow.com/questions/66687797/pyflink-java-io-eofexception-at-java-io-datainputstream-readfully. > You can also find the updated question below: > > I have a PyFlink job that reads from a file, filter based on a condition, > and print. This is a `tree` view of my working directory. This is the > PyFlink script main.py: > > ```python > from pyflink.datastream import StreamExecutionEnvironment > from pyflink.table import StreamTableEnvironment, DataTypes > from pyflink.table.udf import udf > > # https://flink.apache.org/2020/04/09/pyflink-udf-support-flink.html > # > https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/python/table-api-users-guide/udfs/python_udfs.html > > @udf(input_types=[DataTypes.STRING()], > result_type=DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())) > def parse(s): > import json > # a dummy parser > res = {'item_id': 123, 'tag': 'a'} > return res > > env = StreamExecutionEnvironment.get_execution_environment() > t_env = StreamTableEnvironment.create(env) > > t_env.register_function("parse", parse) > > my_source_ddl = """ > create table mySource ( > id BIGINT, > contentstr STRING > ) with ( > 'connector' = 'filesystem', > 'format' = 'json', > 'path' = '/tmp/input' > ) > """ > > my_sink_ddl = """ > create table mySink ( > id BIGINT > ) with ( > 'connector' = 'print' > ) > """ > > my_transform_dml = """ > insert into mySink > with t1 as ( > select id, parse(contentstr) as content > from mySource > ) > select id > from t1 > where content['item_id'] is not null > and content['tag'] = 'a' > """ > > t_env.execute_sql(my_source_ddl) > t_env.execute_sql(my_sink_ddl) > t_env.execute_sql(my_transform_dml).wait() > ``` > > To run the `main.py`: > - Ensure installing pyflink==1.12.0 in my conda env > - /tmp/input has a single row of content `{"id":1,"tag":"a"}` > > Then I run `main.py` and I get the exception: > > ``` > Traceback (most recent call last): > File "udf_parse.py", line 53, in <module> > t_env.execute_sql(my_transform_dml).wait() > File > "/usr/local/anaconda3/envs/pyflink-quickstart/lib/python3.7/site-packages/pyflink/table/table_result.py", > line 76, in wait > get_method(self._j_table_result, "await")() > File > "/usr/local/anaconda3/envs/pyflink-quickstart/lib/python3.7/site-packages/py4j/java_gateway.py", > line 1286, in __call__ > answer, self.gateway_client, self.target_id, self.name) > File > "/usr/local/anaconda3/envs/pyflink-quickstart/lib/python3.7/site-packages/pyflink/util/exceptions.py", > line 147, in deco > return f(*a, **kw) > File > "/usr/local/anaconda3/envs/pyflink-quickstart/lib/python3.7/site-packages/py4j/protocol.py", > line 328, in get_return_value > format(target_id, ".", name), value) > py4j.protocol.Py4JJavaError: An error occurred while calling o53.await. > : java.util.concurrent.ExecutionException: > org.apache.flink.table.api.TableException: Failed to wait job finish > at > java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357) > at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908) > at > org.apache.flink.table.api.internal.TableResultImpl.awaitInternal(TableResultImpl.java:123) > at > org.apache.flink.table.api.internal.TableResultImpl.await(TableResultImpl.java:86) > at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) > at > sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) > at > sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.lang.reflect.Method.invoke(Method.java:498) > at > org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) > at > org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) > at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282) > at > org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) > at > org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79) > at > org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238) > at java.lang.Thread.run(Thread.java:748) > Caused by: org.apache.flink.table.api.TableException: Failed to wait job > finish > at > org.apache.flink.table.api.internal.InsertResultIterator.hasNext(InsertResultIterator.java:56) > at > org.apache.flink.table.api.internal.TableResultImpl$CloseableRowIteratorWrapper.hasNext(TableResultImpl.java:350) > at > org.apache.flink.table.api.internal.TableResultImpl$CloseableRowIteratorWrapper.isFirstRowReady(TableResultImpl.java:363) > at > org.apache.flink.table.api.internal.TableResultImpl.lambda$awaitInternal$1(TableResultImpl.java:110) > at > java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1640) > at > java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) > at > java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) > ... 1 more > Caused by: java.util.concurrent.ExecutionException: > org.apache.flink.runtime.client.JobExecutionException: Job execution failed. > at > java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357) > at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908) > at > org.apache.flink.table.api.internal.InsertResultIterator.hasNext(InsertResultIterator.java:54) > ... 7 more > Caused by: org.apache.flink.runtime.client.JobExecutionException: Job > execution failed. > at > org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144) > at > org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$2(MiniClusterJobClient.java:117) > at > java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616) > at > java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591) > at > java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) > at > java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975) > at > org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$0(AkkaInvocationHandler.java:237) > at > java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774) > at > java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750) > at > java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) > at > java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975) > at > org.apache.flink.runtime.concurrent.FutureUtils$1.onComplete(FutureUtils.java:1046) > at akka.dispatch.OnComplete.internal(Future.scala:264) > at akka.dispatch.OnComplete.internal(Future.scala:261) > at akka.dispatch.japi$CallbackBridge.apply(Future.scala:191) > at akka.dispatch.japi$CallbackBridge.apply(Future.scala:188) > at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36) > at > org.apache.flink.runtime.concurrent.Executors$DirectExecutionContext.execute(Executors.java:73) > at > scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:44) > at > scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:252) > at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:572) > at > akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:22) > at > akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:21) > at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:436) > at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:435) > at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36) > at > akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55) > at > akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:91) > at > akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91) > at > akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91) > at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72) > at > akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:90) > at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40) > at > akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:44) > at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) > at > akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) > at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) > at > akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) > Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed > by NoRestartBackoffTimeStrategy > at > org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:118) > at > org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:80) > at > org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:233) > at > org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:224) > at > org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:215) > at > org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:669) > at > org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:89) > at > org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:447) > at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) > at > sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) > at > sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.lang.reflect.Method.invoke(Method.java:498) > at > org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:305) > at > org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212) > at > org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77) > at > org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158) > at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) > at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) > at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123) > at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) > at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) > at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) > at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) > at akka.actor.Actor$class.aroundReceive(Actor.scala:517) > at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225) > at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592) > at akka.actor.ActorCell.invoke(ActorCell.scala:561) > at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258) > at akka.dispatch.Mailbox.run(Mailbox.scala:225) > at akka.dispatch.Mailbox.exec(Mailbox.scala:235) > ... 4 more > Caused by: java.io.EOFException > at java.io.DataInputStream.readFully(DataInputStream.java:197) > at java.io.DataInputStream.readFully(DataInputStream.java:169) > at > org.apache.flink.table.runtime.typeutils.StringDataSerializer.deserializeInternal(StringDataSerializer.java:88) > at > org.apache.flink.table.runtime.typeutils.StringDataSerializer.deserialize(StringDataSerializer.java:82) > at > org.apache.flink.table.runtime.typeutils.StringDataSerializer.deserialize(StringDataSerializer.java:34) > at > org.apache.flink.table.runtime.typeutils.serializers.python.MapDataSerializer.deserializeInternal(MapDataSerializer.java:129) > at > org.apache.flink.table.runtime.typeutils.serializers.python.MapDataSerializer.deserialize(MapDataSerializer.java:110) > at > org.apache.flink.table.runtime.typeutils.serializers.python.MapDataSerializer.deserialize(MapDataSerializer.java:46) > at > org.apache.flink.table.runtime.typeutils.serializers.python.RowDataSerializer.deserialize(RowDataSerializer.java:106) > at > org.apache.flink.table.runtime.typeutils.serializers.python.RowDataSerializer.deserialize(RowDataSerializer.java:49) > at > org.apache.flink.table.runtime.operators.python.scalar.RowDataPythonScalarFunctionOperator.emitResult(RowDataPythonScalarFunctionOperator.java:81) > at > org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator.emitResults(AbstractPythonFunctionOperator.java:250) > at > org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator.invokeFinishBundle(AbstractPythonFunctionOperator.java:273) > at > org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator.processWatermark(AbstractPythonFunctionOperator.java:199) > at > org.apache.flink.streaming.runtime.tasks.ChainingOutput.emitWatermark(ChainingOutput.java:123) > at > org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask$AsyncDataOutputToOutput.emitWatermark(SourceOperatorStreamTask.java:170) > at > org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask.advanceToEndOfEventTime(SourceOperatorStreamTask.java:110) > at > org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask.afterInvoke(SourceOperatorStreamTask.java:116) > at > org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:589) > at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755) > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570) > at java.lang.Thread.run(Thread.java:748) > ``` > > The issue is probably related to the udf. > > Any help? Thanks! > > Best, > Yik San > > On Fri, Mar 19, 2021 at 11:58 AM Yik San Chan <evan.chanyik...@gmail.com> > wrote: > >> Hi Dian, >> >> I am able to reproduce this issue in a much simpler setup. Let me update >> with the simpler reproducible example shortly. >> >> Best, >> Yik San >> >> On Fri, Mar 19, 2021 at 11:28 AM Yik San Chan <evan.chanyik...@gmail.com> >> wrote: >> >>> Hi Dian, >>> >>> It is a good catch, though after changing to use >>> flink-sql-connector-kafka_2.11-1.12.0.jar I still get exactly the same >>> error. >>> >>> Best, >>> Yik San >>> >>> On Fri, Mar 19, 2021 at 11:02 AM Dian Fu <dian0511...@gmail.com> wrote: >>> >>>> >>>> I noticed that you use "flink-sql-connector-kafka_2.12-1.12.0.jar”. >>>> Does the jar files in the cluster nodes are also built with Scala 2.12? >>>> PyFlink package bundles jar files with Scala 2.11 by default. I’m still not >>>> sure if it’s related to this issue. However, I think this is problematic. >>>> Could you make sure that they are consistent? >>>> >>>> >>>> 2021年3月19日 上午10:40，Yik San Chan <evan.chanyik...@gmail.com> 写道： >>>> >>>> Hi Dian, >>>> >>>> The PyFlink version is 1.12.0 and the Flink version in the cluster >>>> nodes is also 1.12.0 >>>> >>>> $ which flink >>>> /data/apache/flink/flink-1.12.0/bin/flink >>>> >>>> Best, >>>> Yik San >>>> >>>> On Fri, Mar 19, 2021 at 10:26 AM Dian Fu <dian0511...@gmail.com> wrote: >>>> >>>>> Hi, >>>>> >>>>> What’s the Flink version in the cluster nodes? It should matches the >>>>> PyFlink version. >>>>> >>>>> Regards, >>>>> Dian >>>>> >>>>> 2021年3月18日 下午5:01，Yik San Chan <evan.chanyik...@gmail.com> 写道： >>>>> >>>>> This question is cross-posted on StackOverflow >>>>> https://stackoverflow.com/questions/66687797/pyflink-java-io-eofexception-at-java-io-datainputstream-readint >>>>> >>>>> I have a PyFlink job that reads from Kafka source, transform, and >>>>> write to Kafka sink. This is a `tree` view of my working directory. >>>>> >>>>> ``` >>>>> > tree >>>>> . >>>>> ├── deps >>>>> │ └── flink-sql-connector-kafka_2.12-1.12.0.jar >>>>> ├── flink_run.sh >>>>> ├── main.py >>>>> ├── pyflink1.12.0.zip >>>>> └── tasks >>>>> └── user_last_n_clicks >>>>> ├── sink_ddl.sql >>>>> ├── source_ddl.sql >>>>> └── transform_dml.sql >>>>> ``` >>>>> >>>>> This is the `flink_run.sh`: >>>>> >>>>> ``` >>>>> flink run \ >>>>> --yarnname test-pyflink \ >>>>> -m yarn-cluster \ >>>>> -yD yarn.application.queue=tech_platform \ >>>>> -pyarch pyflink1.12.0.zip \ >>>>> -pyexec /data/software/pyflink1.12.0/bin/python \ >>>>> -py main.py testing user_last_n_clicks >>>>> ``` >>>>> >>>>> This is the `main.py`. The key logic is in: >>>>> - `parse_content` udf. >>>>> - load sql files from tasks subfolder, and execute_sql >>>>> >>>>> ```python >>>>> import os >>>>> from sys import argv >>>>> from pyflink.datastream import StreamExecutionEnvironment >>>>> from pyflink.table import StreamTableEnvironment, EnvironmentSettings, >>>>> DataTypes >>>>> from pyflink.table.udf import udf >>>>> >>>>> def read_file_content(filepath): >>>>> with open(filepath) as f: >>>>> return f.read() >>>>> >>>>> @udf(input_types=[DataTypes.STRING()], >>>>> result_type=DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())) >>>>> def parse_content(content_str): >>>>> import json >>>>> res = {} >>>>> content = json.loads(content_str) >>>>> if 'postId' in content: >>>>> res['item_id'] = content['postId'] >>>>> if 'lid' in content: >>>>> res['item_id'] = content['lid'] >>>>> if 'param' in content and 'tag' in content['param']: >>>>> res['tag'] = content['param']['tag'] >>>>> return res >>>>> >>>>> CWD = os.getcwd() >>>>> _, palfish_env, task = argv >>>>> >>>>> VALID_PALFISH_ENVS = ['development', 'testing', 'production'] >>>>> if palfish_env not in VALID_PALFISH_ENVS: >>>>> raise Exception(f"{palfish_env} is not a valid env, should be one >>>>> of [{', '.join(VALID_PALFISH_ENVS)}].") >>>>> >>>>> VALID_TASKS = os.listdir(f"{CWD}/tasks") >>>>> if task not in VALID_TASKS: >>>>> raise Exception(f"{task} is not a valid task, should be one of >>>>> [{', '.join(VALID_TASKS)}].") >>>>> >>>>> config = { >>>>> "development": { >>>>> "${generation.kafka.source.servers}": "localhost:9094", >>>>> "${generation.kafka.sink.servers}": "localhost:9094" >>>>> }, >>>>> "testing": { >>>>> "${generation.kafka.source.servers}": "10.111.135.233:9092, >>>>> 10.111.130.11:9092,10.111.130.12:9092", >>>>> "${generation.kafka.sink.servers}": "10.111.135.233:9092, >>>>> 10.111.130.11:9092,10.111.130.12:9092" >>>>> }, >>>>> "production": { >>>>> "${generation.kafka.source.servers}": "10.111.203.9:9092, >>>>> 10.111.203.10:9092,10.111.203.13:9092,10.111.204.163:9092, >>>>> 10.111.204.164:9092,10.111.204.165:9092", >>>>> "${generation.kafka.sink.servers}": "10.111.209.219:9092, >>>>> 10.111.209.220:9092,10.111.209.221:9092" >>>>> } >>>>> } >>>>> >>>>> FAT_JAR_PATH = f"{CWD}/deps/flink-sql-connector-kafka_2.12-1.12.0.jar" >>>>> >>>>> source_ddl = >>>>> read_file_content(f'{CWD}/tasks/{task}/source_ddl.sql').replace('${generation.kafka.source.servers}', >>>>> config[palfish_env]['${generation.kafka.source.servers}']) >>>>> sink_ddl = >>>>> read_file_content(f'{CWD}/tasks/{task}/sink_ddl.sql').replace('${generation.kafka.sink.servers}', >>>>> config[palfish_env]['${generation.kafka.sink.servers}']) >>>>> transform_dml = >>>>> read_file_content(f'{CWD}/tasks/{task}/transform_dml.sql') >>>>> >>>>> exec_env = StreamExecutionEnvironment.get_execution_environment() >>>>> env_settings = >>>>> EnvironmentSettings.Builder().use_blink_planner().build() >>>>> t_env = >>>>> StreamTableEnvironment.create(stream_execution_environment=exec_env, >>>>> environment_settings=env_settings) >>>>> >>>>> t_env.get_config().get_configuration().set_string("pipeline.jars", >>>>> f"file://{FAT_JAR_PATH}") >>>>> t_env.create_temporary_function("ParseContent", parse_content) >>>>> >>>>> t_env.execute_sql(source_ddl) >>>>> t_env.execute_sql(sink_ddl) >>>>> t_env.execute_sql(transform_dml).wait() >>>>> ``` >>>>> >>>>> See my sqls. Note the udf `ParseContent` is used in >>>>> `transform_dml.sql`. >>>>> >>>>> ```sql >>>>> # source_ddl.sql >>>>> CREATE TABLE kafka_source ( >>>>> `body` ROW<`log` ROW<`uid` BIGINT, serverts BIGINT, `contentstr` >>>>> STRING>> >>>>> ) WITH ( >>>>> 'connector' = 'kafka', >>>>> 'topic' = 'data-report-stat-old-logtype7', >>>>> 'properties.bootstrap.servers' = '${generation.kafka.source.servers}', >>>>> 'properties.group.id' = 'flink-featurepipelines', >>>>> 'format' = 'json' >>>>> ) >>>>> >>>>> # transform_ddl.sql >>>>> INSERT INTO kafka_sink >>>>> WITH t1 AS ( >>>>> SELECT body['log']['uid'] user_id, >>>>> ParseContent(body['log']['contentstr']) content, body['log']['serverts'] >>>>> server_ts >>>>> FROM kafka_source >>>>> ), >>>>> t2 AS ( >>>>> SELECT user_id, content['item_id'] item_id, content['tag'] tag, >>>>> server_ts >>>>> FROM t1 >>>>> WHERE content['item_id'] IS NOT NULL >>>>> AND content['tag'] = '点击帖子卡片' >>>>> ), >>>>> last_n AS ( >>>>> SELECT user_id, item_id, server_ts >>>>> FROM ( >>>>> SELECT *, >>>>> ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY server_ts DESC) as >>>>> row_num >>>>> FROM t2) >>>>> WHERE row_num <= 5 >>>>> ) >>>>> SELECT user_id, CAST(MAX(server_ts / 1000) AS TIMESTAMP) datetime, >>>>> LISTAGG(CAST(item_id AS STRING)) last_5_clicks >>>>> FROM last_n >>>>> GROUP BY user_id >>>>> >>>>> # sink_ddl.sql >>>>> CREATE TABLE kafka_sink ( >>>>> user_id BIGINT, >>>>> datetime TIMESTAMP(3), >>>>> last_5_clicks STRING, >>>>> PRIMARY KEY (user_id) NOT ENFORCED >>>>> ) WITH ( >>>>> 'connector' = 'upsert-kafka', >>>>> 'topic' = 'aiinfra.fct.userfeature.0', >>>>> 'properties.bootstrap.servers' = >>>>> '${generation.kafka.sink.servers}', >>>>> 'key.format' = 'json', >>>>> 'value.format' = 'json' >>>>> ) >>>>> ``` >>>>> >>>>> I got the error when running the PyFlink program in my testing >>>>> environment machine. >>>>> >>>>> ``` >>>>> Caused by: java.io.EOFException >>>>> at java.io.DataInputStream.readInt(DataInputStream.java:392) >>>>> ~[?:1.8.0_261] >>>>> at >>>>> org.apache.flink.table.runtime.typeutils.StringDataSerializer.deserializeInternal(StringDataSerializer.java:91) >>>>> ~[flink-table-blink_2.11-1.12.0.jar:1.12.0] >>>>> at >>>>> org.apache.flink.table.runtime.typeutils.StringDataSerializer.deserialize(StringDataSerializer.java:87) >>>>> ~[flink-table-blink_2.11-1.12.0.jar:1.12.0] >>>>> at >>>>> org.apache.flink.table.runtime.typeutils.StringDataSerializer.deserialize(StringDataSerializer.java:36) >>>>> ~[flink-table-blink_2.11-1.12.0.jar:1.12.0] >>>>> at >>>>> org.apache.flink.table.runtime.typeutils.serializers.python.MapDataSerializer.deserializeInternal(MapDataSerializer.java:124) >>>>> ~[flink-python_2.11-1.12.0.jar:1.12.0] >>>>> at >>>>> org.apache.flink.table.runtime.typeutils.serializers.python.MapDataSerializer.deserialize(MapDataSerializer.java:107) >>>>> ~[flink-python_2.11-1.12.0.jar:1.12.0] >>>>> at >>>>> org.apache.flink.table.runtime.typeutils.serializers.python.MapDataSerializer.deserialize(MapDataSerializer.java:46) >>>>> ~[flink-python_2.11-1.12.0.jar:1.12.0] >>>>> at >>>>> org.apache.flink.table.runtime.typeutils.serializers.python.RowDataSerializer.deserialize(RowDataSerializer.java:104) >>>>> ~[flink-python_2.11-1.12.0.jar:1.12.0] >>>>> at >>>>> org.apache.flink.table.runtime.typeutils.serializers.python.RowDataSerializer.deserialize(RowDataSerializer.java:49) >>>>> ~[flink-python_2.11-1.12.0.jar:1.12.0] >>>>> at >>>>> org.apache.flink.table.runtime.operators.python.scalar.RowDataPythonScalarFunctionOperator.emitResult(RowDataPythonScalarFunctionOperator.java:84) >>>>> ~[flink-python_2.11-1.12.0.jar:1.12.0] >>>>> ``` >>>>> >>>>> Here are the full logs, see >>>>> https://gist.github.com/YikSanChan/d3a5d25cdf2f3c1dc6b3dc93e48c4bbc. >>>>> >>>>> Any idea why the exception? Thanks. >>>>> >>>>> Yik San >>>>> >>>>> >>>>> >>>>
