月份：2019年10月

代码如下

private static voidtest2()throwsException {

    EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();

   StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();

   StreamTableEnvironment tttt = StreamTableEnvironment.create(fsEnv,fsSettings);

   Properties properties =newProperties();

   properties.setProperty("bootstrap.servers","localhost:9092");

    properties.setProperty("group.id","test"+ System.currentTimeMillis());

   FlinkKafkaConsumer011 kafkaConsumer011 =newFlinkKafkaConsumer011("test", newSimpleStringSchema(),properties);

   kafkaConsumer011.setStartFromTimestamp(System.currentTimeMillis());

   DataStream stream = fsEnv.addSource(kafkaConsumer011);

   stream.print();

   fsEnv.execute("test");

}2019-10-21 23:06:58.970 [Thread-8] INFO  o.a.f.s.connectors.kafka.FlinkKafkaConsumerBase - Consumer subtask 0 creating fetcher with offsets {}.

表现为，无法收到kafka发送过来的消息。这就很奇怪了，因为我使用kafka-consumer-console.sh的方式可以看到实际上是有消息产出的，但是这里死活就收不到数据。

问题1： 从postgresql里使用flink-jdbc读取数据的问题, 数据类型不匹配，不支持jsonb等其他类型.

New ElasticSearch()…xxx...registerTableSink();

ElasticSearch是一个ConnectorDescriptor: 最关键的就是一个toConnectorProperties就是一个配置的map,传给factory使用，他其实是一个配置收集器.

目的： 订阅kafka的消息(kafka的消息是从postgresql来的)，将kafka的消息作为TableSource，执行sql，将结果输出到ElasticSearch中

业务代码如下 :

fsTableEnv.connect(

       newKafka()

                .version("0.11")

                .topic("xxxx")

                .property("bootstrap.servers”,”xxxxx:xxxx")

                .property("group.id","test4444")

                .startFromEarliest()//测试需要

)

        .withSchema(newSchema().schema(newTableSchema(fields, types)))

        .withFormat(newJson().schema(newRowTypeInfo(types, fields)))

        .inAppendMode()

        .registerTableSource(“testTable”);

问题是：读出来的数据总是null

DataStream table = fsTableEnv.toAppendStream(fsTableEnv.sqlQuery("select * from testTable"),Row.class);

table.print();

架构

flink是一个框架或者叫流式计算引擎，他可以用来处理有界(批处理，mapReduce)数据和无界数据。他的特点是具有内存的速度和分布式的横向扩展能力。由于是分布式的，所以他可以处理任意规模的数据。另外，他是有状态的，表现为上一步的结算结果可以为下一步使用。