PyFlink里有Row类型的类, pyflink.table.Row
> 在 2020年10月21日，上午9:05，whh_960101 <[email protected]> 写道：
>
> Row类型的对象在python中是怎么表示的，字典？
>
>
>
>
>
>
> 在 2020-10-20 20:35:22，"Dian Fu" <[email protected]> 写道：
>> 你这两个UDF（error_get和headers_get）实际的返回值类型都是string，但是却标成Row类型。如果确实需要返回Row类型，udf的返回值需要返回一个Row类型的对象。
>>
>>> 在 2020年10月20日，下午7:56，Dian Fu <[email protected]> 写道：
>>>
>>> 错误堆栈看着似乎不太完整，有更完整的堆栈吗？
>>>
>>>> 在 2020年10月20日，下午7:38，whh_960101 <[email protected]> 写道：
>>>>
>>>> Hi，各位大佬，我处理kafka source Table，print 一个嵌套的json结果，报错JobExecutionException:
>>>> Job execution failed.
>>>> 这个问题该怎么解决，代码和报错信息如下：@udf(input_types=DataTypes.STRING(),
>>>> result_type=DataTypes.BOOLEAN())
>>>> def error_exist(message):
>>>> if message is None:
>>>> return False
>>>> mes_dic = json.loads(message.strip())
>>>> log = mes_dic.get('log').lower().strip()
>>>> if 'error' in log:
>>>> return True
>>>> else:
>>>> return False
>>>>
>>>> @udf(input_types=DataTypes.STRING(),
>>>> result_type=DataTypes.ROW([DataTypes.FIELD("content",
>>>> DataTypes.STRING())]))
>>>> def error_get(message):
>>>> if message is None:
>>>> return ''
>>>> mes_dic = json.loads(message.strip())
>>>> log = mes_dic.get('log')
>>>> return json.dumps({"content":log.strip()})
>>>>
>>>>
>>>> @udf(input_types=[DataTypes.STRING(),DataTypes.STRING(),DataTypes.STRING()],
>>>> result_type=DataTypes.ROW([\
>>>> DataTypes.FIELD("appId",
>>>> DataTypes.STRING()),DataTypes.FIELD("hostname", DataTypes.STRING()), \
>>>> DataTypes.FIELD("level",
>>>> DataTypes.STRING()),DataTypes.FIELD("timeZone", DataTypes.STRING()), \
>>>> DataTypes.FIELD("timestamp",
>>>> DataTypes.DOUBLE()),DataTypes.FIELD("container", DataTypes.STRING()), \
>>>> DataTypes.FIELD("clusterName",
>>>> DataTypes.STRING()),DataTypes.FIELD("clusterChinese",
>>>> DataTypes.STRING())]))
>>>> def headers_get(message,container,clusterName):
>>>> mes_dic = json.loads(message.strip())
>>>> tz_utc = mes_dic.get('time')
>>>> tz_utc = tz_utc[:tz_utc.rfind('.') + 7]
>>>> from_zone = tz.gettz('UTC')
>>>> dt_utc = datetime.strptime(tz_utc, '%Y-%m-%dT%H:%M:%S.%f')
>>>> dt_utc = dt_utc.replace(tzinfo=from_zone)
>>>> dt_ts = dt_utc.timestamp()
>>>>
>>>> map_df = pd.read_csv('cc_log_dev_map.csv')
>>>> clusterChinese = map_df.loc[map_df.cs_English == clusterName,
>>>> 'cs_Chinese'].values[0]
>>>>
>>>> return
>>>> json.dumps({'appId':'','hostname':'','level':'ERROR','timeZone':'Asia/Shanghai','timestamp':dt_ts,\
>>>>
>>>> 'container':container,'clusterName':clusterName,'clusterChinese':clusterChinese})
>>>>
>>>> #st_env.execute_sql("""
>>>> # CREATE TABLE source(
>>>> # message STRING,
>>>> # clusterName STRING,
>>>> # kubernetes ROW<container ROW<name STRING>>
>>>> # ) WITH(
>>>> # 'connector' = 'kafka',
>>>> # )
>>>> # """)
>>>>
>>>> st_env.execute_sql("""
>>>> CREATE TABLE sink(
>>>> body ROW<content STRING>,
>>>> headers ROW<appId STRING,hostname STRING,level STRING,timeZone
>>>> STRING,\
>>>> `timestamp` DOUBLE,container STRING,clusterName
>>>> STRING,clusterChinese STRING>
>>>> ) WITH(
>>>> 'connector' = 'print',
>>>> )
>>>> """)tmp_table = st_env.from_path("source") \
>>>> .select("message,kubernetes.get('container').get('name') as
>>>> container,clusterName")
>>>>
>>>> data_stream =
>>>> st_env._j_tenv.toAppendStream(tmp_table._j_table,tmp_table._j_table.getSchema().toRowType())
>>>> table =
>>>> Table(st_env._j_tenv.fromDataStream(data_stream,"message,clusterName,container"),st_env)
>>>>
>>>> sink_table = table \
>>>> .where("error_exist(message) = true") \
>>>> .select("error_get(message) as body,
>>>> headers_get(message,container,clusterName) as headers")
>>>>
>>>>
>>>> sink_table.execute_insert("kafka_sink").get_job_client().get_job_execution_result().result()报错：File
>>>> "log_parse_kafka.py", line 180, in from_kafka_to_oracle_demo
>>>>
>>>> sink_table.execute_insert("kafka_sink").get_job_client().get_job_execution_result().result()
>>>> File
>>>> "/home/cdh272705/.local/lib/python3.6/site-packages/pyflink/common/completable_future.py",
>>>> line 78, in result
>>>> return self._py_class(self._j_completable_future.get())
>>>> File
>>>> "/home/cdh272705/.local/lib/python3.6/site-packages/py4j/java_gateway.py",
>>>> line 1286, in __call__
>>>> answer, self.gateway_client, self.target_id, self.name)
>>>> File
>>>> "/home/cdh272705/.local/lib/python3.6/site-packages/pyflink/util/exceptions.py",
>>>> line 147, in deco
>>>> return f(*a, **kw)
>>>> File
>>>> "/home/cdh272705/.local/lib/python3.6/site-packages/py4j/protocol.py",
>>>> line 328, in get_return_value
>>>> format(target_id, ".", name), value)
>>>> py4j.protocol.Py4JJavaError: An error occurred while calling o150.get.
>>>> : 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 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.runtime.client.JobExecutionException: Job
>>>> execution failed.at
>>>> org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:147)
>>>> at
>>>> org.apache.flink.client.program.PerJobMiniClusterFactory$PerJobMiniClusterJobClient.lambda$getJobExecutionResult$2(PerJobMiniClusterFactory.java:186)
>>>> 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:229)
>>>> 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:892)
>>>> 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:74)
>>>> 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)
>>>
