flink中數(shù)據(jù)的落地粹污，是使用sink來(lái)處理的蜕着。

上面例子中已經(jīng)可以看到可以使用DataStream.addSink()方法來(lái)添加數(shù)據(jù)落地的目標(biāo)僵蛛，表示將數(shù)據(jù)輸出到對(duì)應(yīng)目的地互纯。

RichSinkFunction及它的爸爸們：

flink中的sink可以自定義實(shí)現(xiàn)瑟幕，一般需要繼承抽象類(lèi)RichSinkFunction，與數(shù)據(jù)源RichSourceFunction非常類(lèi)似，看下實(shí)現(xiàn)代碼：

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一樣旺聚，這里在之前文章中已經(jīng)看過(guò)榨了。

而主要的方法則是來(lái)自SinkFunction<IN>倒源，IN是一個(gè)輸入泛型，代表需要sink的數(shù)據(jù)類(lèi)型懦鼠∽炅ǎ看下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();
    }
}

可以看到屹堰，這里主要需要關(guān)注的是invoke方法肛冶。

RichSinkFunction的兒子

這里看下自己實(shí)現(xiàn)的mysql數(shù)據(jù)源：

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，用于將數(shù)據(jù)沉淀存儲(chǔ)在不同的位置
 * 這里存儲(chǔ)在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");
    }
}

可以看到扯键，這里主要是重寫(xiě)其中的兩個(gè)方法睦袖，open、invoke荣刑；在open中初始化spring容器馅笙、數(shù)據(jù)庫(kù)鏈接，在invoke中執(zhí)行具體的持久化邏輯厉亏。

使用UrlMySqlSink

數(shù)據(jù)來(lái)源依然來(lái)自kafka董习，復(fù)用之前的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操作爱只，轉(zhuǎn)換皿淋，從一個(gè)數(shù)據(jù)流轉(zhuǎn)換成另一個(gè)數(shù)據(jù)流，這里是從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即可窝趣。一個(gè)datasource可以同時(shí)添加多個(gè)sink。

注意训柴，dataSource是添加到StreamExecutionEnvironment實(shí)例上的哑舒，而sink則是直接添加到dataStreamSource上的。