Apache Flink 漫谈系列(15) - DataStream Connectors之Kafka

Flink Kafka Consumer需要知道如何将Kafka中的二进制数据转换为Java / Scala对象。 DeserializationSchema允许用户指定这样的模式。 为每个Kafka消息调用 T deserialize(byte [] message)方法，从Kafka传递值。

2. Examples

我们示例读取Kafka的数据，再将数据做简单处理之后写入到Kafka中。我们需要再创建一个用于写入的Topic，如下：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic flink-tipic-output 

所以示例中我们Source利用flink-topic, Sink用slink-topic-output。

(1) Simple ETL

我们假设Kafka中存储的就是一个简单的字符串，所以我们需要一个用于对字符串进行serialize和deserialize的实现，也就是我们要定义一个实现DeserializationSchema和SerializationSchema 的序列化和反序列化的类。因为我们示例中是字符串，所以我们自定义一个KafkaMsgSchema实现类,然后在编写Flink主程序。

• KafkaMsgSchema - 完整代码

import org.apache.flink.api.common.serialization.DeserializationSchema; 
import org.apache.flink.api.common.serialization.SerializationSchema; 
import org.apache.flink.api.common.typeinfo.BasicTypeInfo; 
import org.apache.flink.api.common.typeinfo.TypeInformation; 
import org.apache.flink.util.Preconditions; 
 
import java.io.IOException; 
import java.io.ObjectInputStream; 
import java.io.ObjectOutputStream; 
import java.nio.charset.Charset; 
 
public class KafkaMsgSchema implements DeserializationSchema<String>, SerializationSchema<String> { 
    private static final long serialVersionUID = 1L; 
    private transient Charset charset; 
 
    public KafkaMsgSchema() { 
// 默认UTF-8编码 
        this(Charset.forName("UTF-8")); 
    } 
 
    public KafkaMsgSchema(Charset charset) { 
        this.charset = Preconditions.checkNotNull(charset); 
    } 
 
    public Charset getCharset() { 
        return this.charset; 
    } 
 
    public String deserialize(byte[] message) { 
// 将Kafka的消息反序列化为java对象 
        return new String(message, charset); 
    } 
 
    public boolean isEndOfStream(String nextElement) { 
// 流永远不结束 
        return false; 
    } 
 
    public byte[] serialize(String element) { 
// 将java对象序列化为Kafka的消息 
        return element.getBytes(this.charset); 
    } 
 
    public TypeInformation<String> getProducedType() { 
// 定义产生的数据Typeinfo 
        return BasicTypeInfo.STRING_TYPE_INFO; 
    } 
 
    private void writeObject(ObjectOutputStream out) throws IOException { 
        out.defaultWriteObject(); 
        out.writeUTF(this.charset.name()); 
    } 
 
    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { 
        in.defaultReadObject(); 
        String charsetName = in.readUTF(); 
        this.charset = Charset.forName(charsetName); 
    } 
} 

• 主程序 - 完整代码

import org.apache.flink.api.common.functions.MapFunction; 
import org.apache.flink.api.java.utils.ParameterTool; 
import org.apache.flink.streaming.api.datastream.DataStream; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; 
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; 
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper; 
 
import java.util.Properties; 
 
public class KafkaExample { 
    public static void main(String[] args) throws Exception { 
        // 用户参数获取 
        final ParameterTool parameterTool = ParameterTool.fromArgs(args); 
        // Stream 环境 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
 
        // Source的topic 
        String sourceTopic = "flink-topic"; 
        // Sink的topic 
        String sinkTopic = "flink-topic-output"; 
        // broker 地址 
        String broker = "localhost:9092"; 
 
        // 属性参数 - 实际投产可以在命令行传入 
        Properties p = parameterTool.getProperties(); 
        p.putAll(parameterTool.getProperties()); 
        p.put("bootstrap.servers", broker); 
 
        env.getConfig().setGlobalJobParameters(parameterTool); 
 
        // 创建消费者 
        FlinkKafkaConsumer consumer = new FlinkKafkaConsumer<String>( 
                sourceTopic, 
                new KafkaMsgSchema(), 
                p); 
        // 设置读取最早的数据 
//        consumer.setStartFromEarliest(); 
 
        // 读取Kafka消息 
        DataStream<String> input = env.addSource(consumer); 
 
 
        // 数据处理 
        DataStream<String> result = input.map(new MapFunction<String, String>() { 
            public String map(String s) throws Exception { 
                String msg = "Flink study ".concat(s); 
                System.out.println(msg); 
                return msg; 
            } 
        }); 
 
        // 创建生产者 
        FlinkKafkaProducer producer = new FlinkKafkaProducer<String>( 
                sinkTopic, 
                new KeyedSerializationSchemaWrapper<String>(new KafkaMsgSchema()), 
                p, 
                FlinkKafkaProducer.Semantic.AT_LEAST_ONCE); 
 
        // 将数据写入Kafka指定Topic中 
        result.addSink(producer); 
 
        // 执行job 
        env.execute("Kafka Example"); 
    } 
} 

（编辑：西安站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!