flink学习之五-数据持久化to-mysql

flink中数据的落地，是使用sink来处理的。

上面例子中已经可以看到可以使用DataStream.addSink()方法来添加数据落地的目标，表示将数据输出到对应目的地。

RichSinkFunction及它的爸爸们：

flink中的sink可以自定义实现，一般需要继承抽象类RichSinkFunction，与数据源RichSourceFunction非常类似，看下实现代码：

package org.apache.flink.streaming.api.functions.sink;

import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.functions.AbstractRichFunction;

@Public
public abstract class RichSinkFunction<IN> extends AbstractRichFunction implements SinkFunction<IN> {
    private static final long serialVersionUID = 1L;

    public RichSinkFunction() {
    }
}

可以看下，依然是继承AbstractRichFunction，跟RichSourceFunction一样，这里在之前文章中已经看过。

而主要的方法则是来自SinkFunction<IN>，IN是一个输入泛型，代表需要sink的数据类型。看下SinkFunction的定义：

package org.apache.flink.streaming.api.functions.sink;

import java.io.Serializable;
import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.functions.Function;

@Public
public interface SinkFunction<IN> extends Function, Serializable {
    /** @deprecated */
    @Deprecated
    default void invoke(IN value) throws Exception {
    }

    default void invoke(IN value, SinkFunction.Context context) throws Exception {
        this.invoke(value);
    }

    @Public
    public interface Context<T> {
        long currentProcessingTime();

        long currentWatermark();

        Long timestamp();
    }
}

可以看到，这里主要需要关注的是invoke方法。

RichSinkFunction的儿子

这里看下自己实现的mysql数据源：

package myflink.sinks;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import myflink.manager.UrlInfoManager;
import myflink.model.UrlInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.support.ClassPathXmlApplicationContext;

/**
 * sink，用于将数据沉淀存储在不同的位置
 * 这里存储在mysql中的url_info表
 */
@Slf4j
@NoArgsConstructor
@AllArgsConstructor
public class UrlMysqlSink extends RichSinkFunction<UrlInfo> implements ApplicationContextAware {

    private UrlInfoManager urlInfoManager;

    private ApplicationContext applicationContext;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        log.info("applicationContext=" + applicationContext);
        if (applicationContext == null) {
            init();
        }
    }

    @Override
    public void invoke(UrlInfo value, Context context) throws Exception {
        if (urlInfoManager == null) {
            init();
        }
        urlInfoManager.insert(value);
        log.info("---insert url info:", JSON.toJSONString(value));
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    private void init() {
        applicationContext = new ClassPathXmlApplicationContext("classpath*:applicationContext.xml");
        urlInfoManager = (UrlInfoManager) applicationContext.getBean("urlInfoManager");
    }
}

可以看到，这里主要是重写其中的两个方法，open、invoke；在open中初始化spring容器、数据库链接，在invoke中执行具体的持久化逻辑。

使用UrlMySqlSink

数据来源依然来自kafka，复用之前的kafkaSender。

import java.util.Properties;

@Slf4j
public class KafkaUrlSinkJob {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("zookeeper.connect", "localhost:2181");
        properties.put("group.id", "metric-group");
        properties.put("auto.offset.reset", "latest");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        SingleOutputStreamOperator<UrlInfo> dataStreamSource = env.addSource(
                new FlinkKafkaConsumer010<String>(
                        "testjin",// topic
                        new SimpleStringSchema(),
                        properties
                )
        ).setParallelism(1)
                // map操作，转换，从一个数据流转换成另一个数据流，这里是从string-->UrlInfo
                .map(string -> JSON.parseObject(string, UrlInfo.class))；
            
       dataStreamSource.addSink(new UrlMysqlSink());
       dataStreamSource.addSink(new PrintSinkFunction<>());

        env.execute("save url to db");

直接在datasource中addSink即可。一个datasource可以同时添加多个sink。

注意，dataSource是添加到StreamExecutionEnvironment实例上的，而sink则是直接添加到dataStreamSource上的。