Exception "Table of atomic type can only have a single field." when convert DataStream to table

Question

1、BillCount and Record are class objects. BillCount object's columns are some of Record's.
2、Flink source is getting 'Record' data from kafka topic.

case  class BillCount(logisId: Int, provinceId: Int, cityId: Int, orderRequVari: Int, orderRecAmount: Double, orderRecDate: Long)
val kafkaInputStream: DataStream[Record] = env.addSource(source)   //source is FlinkKafkaConsumer010 source
   val tbDataStream : DataStream[BillCount] = kafkaInputStream.map(
              new MapFunction[Record, BillCount] {
                override def map(value: Record) = {
                  BillCount(value.getLogis_id, value.getProvince_id, value.getCity_id,
                          value.getOrder_require_varieties, value.getOrder_rec_amount, value.getStore_rec_date.getTime)
    }
  })
 val stream = tbDataStream.toTable(tbEnv, 'logisId, 'provinceId, 'cityId, 'orderRequVari, 'orderRecAmount, 'orderRecDate) // occur error here

Following is exception :

Exception in thread "main" org.apache.flink.table.api.TableException: Table of atomic type can only have a single field.
at org.apache.flink.table.api.TableEnvironment$$anonfun$1.apply(TableEnvironment.scala:627)
at org.apache.flink.table.api.TableEnvironment$$anonfun$1.apply(TableEnvironment.scala:624)
at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:251)
at scala.collection.mutable.ArrayOps$ofRef.flatMap(ArrayOps.scala:108)
at org.apache.flink.table.api.TableEnvironment.getFieldInfo(TableEnvironment.scala:624)
at org.apache.flink.table.api.StreamTableEnvironment.registerDataStreamInternal(StreamTableEnvironment.scala:398)
at org.apache.flink.table.api.scala.StreamTableEnvironment.fromDataStream(StreamTableEnvironment.scala:85)
at org.apache.flink.table.api.scala.DataStreamConversions.toTable(DataStreamConversions.scala:58)

Can you post the type of your stream that you are trying to convert? println(tbDataStream.dataType) I think the problem is that BillCount is treated as a generic type, because you don't meet the POJO requirements. — twalthr
Usually, the log tells you why BillCount is not a POJO. Btw. it must be Java static (defined in a Scala companion object). — twalthr
Thanks for your answer sincerely. After recreating BillCount class, println(tbDataStream.dataType) 's result is PojoType<org.apache.flink.app.BillCount, fields = [cityId: Integer, logisId: Integer,...> , not occuring previous error log. :) Thanks again. — Li Peng

pijing pijing · Accepted Answer · 2018-10-23T04:00:37

BillCount should be a POJO class. It means that BillCount class should has a default constructor(no parameter) and get/set functions. exp:

class BillCount{
    private int logisId;
    private int provinceId;
    private int cityId;

    BillCount(){}

    public void setLogisId(int logisId){
        this.logisId = logisId;
    }

    public void setProvinceId(int provinceId){
        this.provinceId = provinceId;
    }

    public void setCityId(int cityId){
        this.cityId = cityId;
    }

    public int getLogisId(){
        return this.logisId;
    }

    public int getProvinceId(){
        return this.provinceId;
    }

    public int getCityId() {
        return this.cityId;
    }

}

Exception "Table of atomic type can only have a single field." when convert DataStream to table

1 Answers