非常感谢, 我用的flink-1.9.2 , 但是直接将代码copy过来可以用了!

Jingsong Li <jingsongl...@gmail.com> 于2020年4月24日周五 下午3:02写道:

> 1.10里面有TableUtils了,里面有collectToList
>
>
> Best,
> Jingsong Lee
>
> On Fri, Apr 24, 2020 at 2:49 PM jun su <sujun891...@gmail.com> wrote:
>
> > hi all,
> >
> > 找到了源码中BatchTableEnvUtil类使用了CollectTableSink来做collect的逻辑,
> > 但是下方代码运用了源码内部的private方法, 看起来不允许外部调用:
> >
> > def collect[T](
> >     tEnv: TableEnvironment,
> >     table: Table,
> >     sink: CollectTableSink[T],
> >     jobName: Option[String]): Seq[T] = {
> >   val typeSerializer = fromDataTypeToLegacyInfo(sink.getConsumedDataType)
> >     .asInstanceOf[TypeInformation[T]]
> >     .createSerializer(tEnv.asInstanceOf[TableEnvironmentImpl]
> >       .getPlanner.asInstanceOf[PlannerBase].getExecEnv.getConfig)
> >   val id = new AbstractID().toString
> >   sink.init(typeSerializer.asInstanceOf[TypeSerializer[T]], id)
> >   val sinkName = UUID.randomUUID().toString
> >   tEnv.registerTableSink(sinkName, sink)
> >   tEnv.insertInto(table, sinkName)
> >
> >   val res = tEnv.execute("test")
> >   val accResult: JArrayList[Array[Byte]] = res.getAccumulatorResult(id)
> >   SerializedListAccumulator.deserializeList(accResult, typeSerializer)
> > }
> >
> >
> > jun su <sujun891...@gmail.com> 于2020年4月24日周五 下午2:05写道:
> >
> > > hi all,
> > >
> > > blink模式下没法将table 转为 dataset , 所以如果直接collect了, 请问有类似方法可以获取到
> > > 结果用于代码调试么?
> > >
> > > --
> > > Best,
> > > Jun Su
> > >
> >
> >
> > --
> > Best,
> > Jun Su
> >
>
>
> --
> Best, Jingsong Lee
>


-- 
Best,
Jun Su

回复