你这两个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) >
