Flink读取Kafka数据

Streams can transport data between two operators in a one-to-one(一对一) 
(or forwarding) pattern, or in a redistributing(重新分发) pattern:
One-to-one streams (for example between the Source and the map() operators 
in the figure above) preserve the partitioning and ordering of the 
elements.(保持分区和排序) That means that subtask[1] of the map() operator 
will see the same elements in the same order as they were produced by 
subtask[1] of the Source operator.


Redistributing streams (as between map() and keyBy/window above, 
as well as between keyBy/window and Sink) change the partitioning 
of streams(改变分区). Each operator subtask sends data to different 
target subtasks, depending on the selected transformation. Examples are 
keyBy() (which re-partitions by hashing the key), broadcast(), 
or rebalance() (which re-partitions randomly). In a redistributing 
exchange the ordering among the elements is only preserved within 
each pair of sending and receiving subtasks(只在对应的发送方和接收方保持排序) 
(for example, subtask[1] of map() and subtask[2] of keyBy/window). 
So in this example, the ordering within each key is preserved, 
but the parallelism(并行度) does introduce non-determinism regarding the order 
in which the aggregated results for different keys arrive at the sink.

When referring to time in a streaming program (for example to 
define windows), one can refer to different notions of time:
Event Time is the time when an event was created. It is usually 
described by a timestamp in the events, for example attached by 
the producing sensor, or the producing service. Flink accesses 
event timestamps via timestamp assigners.
Ingestion time is the time when an event enters the Flink dataflow 
at the source operator.
Processing Time is the local time at each operator that performs a 
time-based operation.

环境配置
192.168.127.128：hadoop
192.168.127.129:kafka(kafka_2.11-0.10.2.1)

启动hadoop
cd到hadoop的sbin目录执行 ./start-all.sh

启动zookeeper
cd到zookeeper的bin目录执行 ./zkServer.sh start

启动kafka
./kafka-server-start.sh ../config/server.properties
查看kafka topic列表
./kafka-topics.sh --list --zookeeper localhost:2181
创建topic
./kafka-topics.sh --create --topic topicname --replication-factor 1 --partitions 1 --zookeeper localhost:2181
删除topic
./kafka-topics.sh  --delete --zookeeper localhost:2181  --topic test2
Topic test2 is marked for deletion
此时topic只是标记被删除，并没有真正的删除，真正的删除要到zookeeper目录下删除
cd到zookeeper的bin目录下执行 ./zkCli.sh，进入zookeeper命令行
执行 ls /brokers/topics 可以看到所有的topic路径
[test2, con, test, test1, __consumer_offsets]
找到要删除的topic路径，执行 rmr /brokers/topics/test2
再执行 ls /brokers/topics
[con, test, test1, __consumer_offsets]
可以看到test2已经被删除了
再进到kafka/bin目录下执行./kafka-topics.sh --list --zookeeper localhost:2181
__consumer_offsets
con
test
test1 可以看到test2已经被删除了

package com.cesgroup.example.kafka;


import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.util.Collector;


import java.util.Properties;


/**
 *
 */
public class KafkaMessageStreaming {
    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "192.168.127.129:9092");
        props.setProperty("group.id", "test-consumer-group");


        FlinkKafkaConsumer010<String> consumer =
                new FlinkKafkaConsumer010<String>("test1", new SimpleStringSchema(), props);
        consumer.assignTimestampsAndWatermarks(new MessageWaterEmitter());


        DataStream<Tuple2<String, Long>> kafkaStream = env
                .addSource(consumer)
                .flatMap(new MessageSplitter())
                .keyBy(0)
                .timeWindow(Time.seconds(10))


                .apply(new WindowFunction<Tuple2<String, Long>, Tuple2<String, Long>, Tuple, TimeWindow>() {
                    @Override
                    public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<Tuple2<String, Long>> out) throws Exception {
                        long sum = 0L;
                        int count = 0;
                        for (Tuple2<String, Long> record: input) {
                            sum += record.f1;
                            count++;
                        }
                        Tuple2<String, Long> result = input.iterator().next();
                        result.f1 = sum / count;
                        out.collect(result);
                    }
                });


        kafkaStream.writeAsText("/opt/result.txt", FileSystem.WriteMode.OVERWRITE);
        kafkaStream.print();
        env.execute("Flink-Kafka demo");


    }
}

package com.cesgroup.example.kafka;


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;


/**
 * split message
 */
public class MessageSplitter implements FlatMapFunction<String, Tuple2<String, Long>> {


    @Override
    public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
        if (value != null && value.contains(",")) {
            String[] parts = value.split(",");
            out.collect(new Tuple2<String, Long>(parts[1], Long.parseLong(parts[2])));
        }
    }
}

package com.cesgroup.example.kafka;




import com.sun.management.OperatingSystemMXBean;


import java.lang.management.ManagementFactory;


public class MemoryUsageExtrator {


    private static OperatingSystemMXBean mxBean =
            (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean();


    /**
     * Get current free memory size in bytes
     * @return  free RAM size
     */
    public static long currentFreeMemorySizeInBytes() {
        return mxBean.getFreePhysicalMemorySize();
    }
}

package com.cesgroup.example.kafka;


import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;


import javax.annotation.Nullable;


/**
 * watermark
 */
public class MessageWaterEmitter implements AssignerWithPunctuatedWatermarks<String> {
    @Nullable
    @Override
    public Watermark checkAndGetNextWatermark(String lastElement, long extractedTimestamp) {
        if (lastElement != null && lastElement.contains(",")) {
            String[] parts = lastElement.split(",");
            return new Watermark(Long.parseLong(parts[0]));
        }
        return null;
    }


    @Override
    public long extractTimestamp(String element, long previousElementTimestamp) {
        if (element != null && element.contains(",")) {
            String[] parts = element.split(",");
            return Long.parseLong(parts[0]);
        }
        return 0L;
    }
}