flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ChenDonghe (JIRA)" <j...@apache.org>
Subject [jira] [Created] (FLINK-9025) Problem with convert String to Row
Date Mon, 19 Mar 2018 16:34:00 GMT
ChenDonghe created FLINK-9025:

             Summary: Problem with convert String to Row
                 Key: FLINK-9025
                 URL: https://issues.apache.org/jira/browse/FLINK-9025
             Project: Flink
          Issue Type: Bug
          Components: Java API
            Reporter: ChenDonghe

when we convert json string into Row, and set type of filed in the row, code below:
// code placeholder
DataStream<String> stream = env.addSource(consumer.getInstance(sourceTopic, new SimpleStringSchema()).setStartFromLatest());

DataStream<Row> dataStreamRow = stream.map(new ConvertDataStream()).returns(typeinfo);
this convert process calls the copy function in RowSerializer class, code below:

// public final class RowSerializer extends TypeSerializer<Row>

public Row copy(Row from) {
   int len = fieldSerializers.length;

   if (from.getArity() != len) {
      throw new RuntimeException("Row arity of from does not match serializers.");

   Row result = new Row(len);
   for (int i = 0; i < len; i++) {
      Object fromField = from.getField(i);
      if (fromField != null) {
         Object copy = fieldSerializers[i].copy(fromField);
         result.setField(i, copy);
      } else {
         result.setField(i, null);
   return result;
the json string type message from kafka convert to the row type, in this process, RowSerrializer
copy the from-row to a new result-row, but the type of result-row filed is Object type, for
example, the first message from kafka filed0 is Integer type, the second message from kafka
filed0 is Long type, if we set the filed0 is Long, we hope Integer can be compatible, but
we got an exception : can not convert Interger to Long, so we hope RowSerrializer copy function
can be more flexible, it can be act as a table, we can insert an Integer type value into Long
type filed.


This message was sent by Atlassian JIRA

View raw message