即使在阿帕奇弗林克

Question

我已经写了简单的样品甚至实现AppendTableSink接口后下沉表，但得到这个例外在Apache中弗林克实施AppendTableStream接口后获得下面的错误。

package com.cc.flink.functionUtils; 

    import java.io.IOException; 
    import java.util.ArrayList; 
    import java.util.Collection; 
    import java.util.Iterator; 

    import org.apache.flink.api.common.functions.IterationRuntimeContext; 
    import org.apache.flink.api.common.functions.MapFunction; 
    import org.apache.flink.api.common.functions.RichFunction; 
    import org.apache.flink.api.common.io.OutputFormat; 
    import org.apache.flink.api.common.typeinfo.TypeInformation; 
    import org.apache.flink.api.java.io.LocalCollectionOutputFormat; 
    import org.apache.flink.api.java.tuple.Tuple2; 
    import org.apache.flink.api.java.typeutils.TupleTypeInfo; 
    import org.apache.flink.configuration.Configuration; 
    import org.apache.flink.contrib.streaming.DataStreamUtils; 
    import org.apache.flink.streaming.api.datastream.DataStream; 
    import org.apache.flink.streaming.api.datastream.DataStreamSink; 
    import org.apache.flink.streaming.api.datastream.DataStreamSource; 
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
    import org.apache.flink.streaming.api.functions.sink.SinkFunction; 
    import org.apache.flink.streaming.connectors.rabbitmq.RMQSource; 
    import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig; 
    import org.apache.flink.streaming.util.serialization.SimpleStringSchema; 
    import org.apache.flink.table.api.Table; 
    import org.apache.flink.table.api.TableEnvironment; 
    import org.apache.flink.table.api.java.StreamTableEnvironment; 
    import org.apache.flink.table.sinks.AppendStreamTableSink; 
    import org.apache.flink.table.sinks.RetractStreamTableSink; 
    import org.apache.flink.table.sinks.TableSink; 
    import org.apache.flink.types.Row; 




    public class MyTable implements AppendStreamTableSink<Row>{ 



     @Override 
     public TableSink<Row> configure(String[] arg0, TypeInformation<?>[] arg1) { 
      // TODO Auto-generated method stub 
      return null; 
     } 

     @Override 
     public String[] getFieldNames() { 
      // TODO Auto-generated method stub 
      return null; 
     } 

     @Override 
     public TypeInformation<?>[] getFieldTypes() { 
      // TODO Auto-generated method stub 
      return null; 
     } 

     @Override 
     public TypeInformation<Row> getOutputType() { 
      // TODO Auto-generated method stub 
      return null; 
     } 

     @Override 
     public void emitDataStream(DataStream<Row> arg0) { 
      // TODO Auto-generated method stub 
      arg0.print(); 

     } 




     public static void main(String[] args) throws Exception { 

      final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
      final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder() 
        .setHost("localhost") 
        .setVirtualHost("/") 
        .setUserName("guest") 
        .setPassword("guest") 
        .setPort(5672) 
        .build(); 


      final DataStream<String> stream = env 
        .addSource(new RMQSource<String>(
          connectionConfig,   // config for the RabbitMQ connection 
          "test",     // name of the RabbitMQ queue to consume 
          true,      // use correlation ids; can be false if only at-least-once is required 
          new SimpleStringSchema())) // deserialization schema to turn messages into Java objects 
        .setParallelism(1); 



      final ArrayList<String> values = new ArrayList<>(); 
      StreamTableEnvironment StreamTableEnv = TableEnvironment.getTableEnvironment(env); 
      Table fromDataStream = StreamTableEnv.fromDataStream(stream, 
        "member_id"); 
      StreamTableEnv.registerTable("emp1",fromDataStream); 
      Table output =StreamTableEnv.sql("select count(*) from emp1 where member_id Like '%test%'"); 
      fromDataStream.writeToSink(new MyTable()); 
      env.execute(); 

     } 

    }  

的log4j：警告没有附加目的地可以为记录器（org.apache.calcite.sql.parser）中找到。 log4j：WARN请正确初始化log4j系统。 的log4j：WARN看http://logging.apache.org/log4j/1.2/faq.html#noconfig在线程的详细信息

异常 “主要” org.apache.flink.table.api.TableException：流表只能由AppendStreamTableSink发出，RetractStreamTable

在org.apache。 flink.table.api.StreamTableenvironment.writeToSink（StreamTableenvironment.scala：219）

在org.apache.flink.table.api.Table.writeToSink（table.scala：800）

在org.apache。 flink.table.api.Table.writeToSink（table.scala：773）

在com.cc.flink.functionutils.MyTable.main（MyTable.java:103）

Answer 1

在你的榜样的问题是，你要使用的AppendTableSink但你的查询产生的撤稿。这是因为在你的语句COUNT(*)。每当新行到达时，旧的发射计数不再有效，需要收回。

如果这纯粹是一个SELECT *，比每传入行会产生只有一个输出行不影响以前行。