Flink的Sink（kafka，mysql , redis）

你家宝宝

1629人浏览 · 2019-12-09 10:03:01

你家宝宝 · 2019-12-09 10:03:01 发布

Flink的内嵌Sink

writeAsText ：保存为文本
writeAsCsv ：保存为csv格式文本
print 和printToErr：打印输出（标准输出和错误输出）
writeUsingOutputFormat：自定义格式输出（text和csv都是调用了它）
writeToSocket：输出到某一端口

自定义连接

1. 所需要的pom依赖

<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<flink.version>1.9.0</flink.version>
		<java.version>1.8</java.version>
		<scala.binary.version>2.11</scala.binary.version>
		<maven.compiler.source>${java.version}</maven.compiler.source>
		<maven.compiler.target>${java.version}</maven.compiler.target>
	</properties>

<dependencies>
		<!-- Apache Flink dependencies -->
		<!-- These dependencies are provided, because they should not be packaged into the JAR file. -->
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-java</artifactId>
			<version>${flink.version}</version>
		</dependency>
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
			<version>${flink.version}</version>
		</dependency>
		<dependency>
			<groupId>org.projectlombok</groupId>
			<artifactId>lombok</artifactId>
			<version>1.16.18</version>
		</dependency>

		<!-- Add connector dependencies here. They must be in the default scope (compile). -->

		<!-- Example:

		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-connector-kafka-0.10_${scala.binary.version}</artifactId>
			<version>${flink.version}</version>
		</dependency>
		-->

		<!-- Add logging framework, to produce console output when running in the IDE. -->
		<!-- These dependencies are excluded from the application JAR by default. -->
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-log4j12</artifactId>
			<version>1.7.7</version>
		</dependency>
		<dependency>
			<groupId>log4j</groupId>
			<artifactId>log4j</artifactId>
			<version>1.2.17</version>
		</dependency>

		<!--连接kafka-->
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
			<version>1.9.0</version>
		</dependency>
		<!--连接mysql-->
		<dependency>
			<groupId>mysql</groupId>
			<artifactId>mysql-connector-java</artifactId>
			<version>8.0.16</version>
		</dependency>

		<dependency>
			<groupId>org.apache.commons</groupId>
			<artifactId>commons-pool2</artifactId>
			<version>2.7.0</version>
		</dependency>
		<!--连接redis-->
		<dependency>
			<groupId>com.github.sazzad16</groupId>
			<artifactId>jedis</artifactId>
			<version>2.9.2</version>
		</dependency>
	</dependencies>

2. 本机环境

mysql 8.x版本
kafka2.2.x版本
idea 2018年12月版本
jdk1.8版本
安装了lombok

3 .kafka的Sink

package org.feng.sink;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
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.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

/**
 * Created by Feng on 2019/12/6 16:57
 * CurrentProject's name is flink
 * Flink输出到（Sink）Kafka
 * 过程：消费kafka(in-topic)-计算-写入数据到kafka(out-topic)
 * @author Feng
 */
public class KafkaSink {

    /**
     * kafka需要的属性：zookeeper的位置
     */
    private static Properties properties;
    static {
        properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
    }

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

        // 从最开始接收Kafka上的数据
        DataStream<String> dataStream = env.addSource(getConsumer());

        // 定义计算结果到 ProducerRecord 的转换
        // 这里可以使用 Lambda 表达式代替
        KafkaSerializationSchema<String> kafkaSerializationSchema = new KafkaSerializationSchema<String>() {

            private static final long serialVersionUID = 9079854407359463975L;

            @Override
            public ProducerRecord<byte[], byte[]> serialize(String s, @Nullable Long aLong) {
                return new ProducerRecord<>("out-topic", s.getBytes());
            }
        };

        // 定义Flink Kafka生产者：发送消息为精准一次
        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>("out-topic",
                kafkaSerializationSchema, properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        // 字符串转小写；
        // 写入Kafka
        dataStream.map(String::toLowerCase).addSink(kafkaProducer);

        env.execute("Flink to Kafka");
    }


    /**
     * 获取Flink-Kafka消费者对象
     * @return FlinkKafkaConsumer
     */
    private static FlinkKafkaConsumer<String> getConsumer(){
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("in-topic", new SimpleStringSchema(), properties);
        consumer.setStartFromEarliest();
        return consumer;
    }
}

4. mysql的Sink

4.1 数据库表

user表

4.2 实体类

package org.feng.sink;

import lombok.AllArgsConstructor;
import lombok.Data;

/**
 * Created by Feng on 2019/12/9 8:47
 * CurrentProject's name is flink
 * @author Feng
 */
@Data
@AllArgsConstructor
public class User {

    private Integer id;
    private String name;
    private String password;
    private Long createDate;

    public User(String name){
        this.name = name;
    }
}

4.3 Sink类

package org.feng.sink;

import com.mysql.cj.jdbc.Driver;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * Created by Feng on 2019/12/6 17:32
 * CurrentProject's name is flink
 * @author Feng
 */
public class MysqlSink extends RichSinkFunction<User> {

    private static final long serialVersionUID = 1421283325833689633L;

    private PreparedStatement pstm;
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 加载驱动
        new Driver();

        // 获取连接
        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/demo2019?characterEncoding=UTF-8" +
                "&serverTimezone=UTC", "root", "root");
        // 创建语句
        String sql = "insert into user (name, password, create_date) values (?,?,?)";
        pstm = connection.prepareStatement(sql);
    }

    @Override
    public void invoke(User user, Context context) throws Exception {
        pstm.setString(1, user.getName());
        pstm.setString(2, user.getPassword());
        pstm.setLong(3, user.getCreateDate());

        pstm.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        super.close();
        if(pstm != null){
            pstm.close();
        }

        if(connection != null){
            connection.close();
        }
    }
}

4.4 Sink客户端测试

package org.feng.sink;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Feng on 2019/12/9 8:58
 * CurrentProject's name is flink
 * @author Feng
 */
public class MysqlSinkClient {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<User> userDataStream =
                env.fromElements(new User(null, "feng", "12345",
                System.currentTimeMillis()), new User(null, "liqian",
                "123456", System.currentTimeMillis()));

        // 加入自定义Sink
        userDataStream.addSink(new MysqlSink());

        env.execute();
    }

}

4.5 结果查看

在这里插入图片描述
数据插入成功了。

5. Redis 的Sink

注意：这里用的User类是mysql测试时用的那个User类

5.1 Sink类

package org.feng.sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

/**
 * Created by Feng on 2019/12/9 9:08
 * CurrentProject's name is flink
 * @author Feng
 */
public class RedisSink extends RichSinkFunction<User> {
    private static final long serialVersionUID = -5127825797809677657L;

    private Jedis jedis;
    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = new Jedis("localhost", 6379);
    }

    @Override
    public void invoke(User user, Context context) throws Exception {
        jedis.sadd("userSet", user.getName());
    }

    @Override
    public void close() throws Exception {
        if(jedis != null){
            jedis.close();
        }
    }
}

5.2 客户端测试类

package org.feng.sink;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import redis.clients.jedis.Jedis;

/**
 * Created by Feng on 2019/12/9 9:27
 * CurrentProject's name is flink
 * @author Feng
 */
public class RedisSinkClient {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<User> userDataStream = env.fromElements(new User("fengjinsong1"), new User("liming"));
        userDataStream.addSink(new RedisSink());
        env.execute();

        Jedis jedis = new Jedis("localhost", 6379);
        // 打印插入set的结果
        jedis.smembers("userSet").forEach(System.out::println);
    }
}

5.3 测试结果

给redis中的set的存储数据，不会重复。

fengjinsong
liming
fengjinsong1

Process finished with exit code 0

6. Flink-Redis的整合包使用

6.1 导入依赖(在前边的基础上加上这个依赖)：

<!--flink自己整合的redis-->
   	<dependency>
   		<groupId>org.apache.bahir</groupId>
   		<artifactId>flink-connector-redis_2.11</artifactId>
   		<version>1.0</version>
   	</dependency>

6.2 java 代码

package org.feng.redis;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import redis.clients.jedis.Jedis;

/**
 * Created by Feng on 2019/12/9 10:58
 * CurrentProject's name is flink
 * @author Feng
 */
public class RedisSinkDemo {

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

        FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder().setHost("localhost").build();

        DataStream<String> dataStream = env.fromElements("feng", "li", "zhao", "liu", "hu");
        // 创建RedisSink对象
        dataStream.map((MapFunction<String, Tuple2<String, String>>) value ->
             new Tuple2<>("userSets", value)
        ).returns(Types.TUPLE(Types.STRING, Types.STRING))
                .addSink(new RedisSink<>(config, new RedisMapperImpl()));

        // 打印结果
        Jedis jedis = new Jedis("localhost", 6379);
        jedis.smembers("userSets").forEach(System.out::println);
        env.execute("RedisSinkDemo");
    }


    /**
     * 设置Redis key并将flink数据类型映射到redis数据类型
     */
    private static class RedisMapperImpl implements RedisMapper<Tuple2<String, String>> {

        private static final long serialVersionUID = -8714885560557692676L;

        /**
         * 设置数据使用的数据结构为 set；即redis中命令为sadd
         * @return RedisCommandDescription
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.SADD, "userSets");
        }

        /**
         * 映射key
         * @param stringLongTuple2 数据
         * @return key
         */
        @Override
        public String getKeyFromData(Tuple2<String, String> stringLongTuple2) {
            return stringLongTuple2.f0;
        }

        /**
         * 映射value
         * @param stringLongTuple2 数据
         * @return value
         */
        @Override
        public String getValueFromData(Tuple2<String, String> stringLongTuple2) {
            return stringLongTuple2.f1;
        }
    }

}

6.3 测试结果

控制台输出：

feng
li
liu
zhao
hu
11:45:17,045 WARN  org.apache.flink.runtime.webmonitor.WebMonitorUtils           - Log file environment variable 'log.file' is not set.
11:45:17,046 WARN  org.apache.flink.runtime.webmonitor.WebMonitorUtils           - JobManager log files are unavailable in the web dashboard. Log file location not found in environment variable 'log.file' or configuration key 'Key: 'web.log.path' , default: null (fallback keys: [{key=jobmanager.web.log.path, isDeprecated=true}])'.

Process finished with exit code 0

redis客户端：

redis 127.0.0.1:6379> smembers userSets
1) "feng"
2) "li"
3) "liu"
4) "zhao"
5) "hu"

乐奇 Rokid 开放社区

更多推荐

从源码拆解一个 AIUI 每日一词智能体：页面、离线索引与硬件交互

乐奇 Rokid 开放社区

一句话打造「丢哪里」垃圾分类AIUI智能体

乐奇 Rokid 开放社区

Rokid Glasses 2 的三层补能系统：263mAh 眼镜电池、胶囊电池与充电盒，如何转化为可量产方案？

乐奇 Rokid 开放社区

所有评论(0)

查看更多评论

你家宝宝

@FBB360JAVA

已为社区贡献2条内容

Flink的Sink（kafka，mysql , redis）

你家宝宝

Flink的内嵌Sink

自定义连接

1. 所需要的pom依赖

2. 本机环境

3 .kafka的Sink

4. mysql的Sink

4.1 数据库表

4.2 实体类

4.3 Sink类

4.4 Sink客户端测试

4.5 结果查看

5. Redis 的Sink

5.1 Sink类

5.2 客户端测试类

5.3 测试结果

6. Flink-Redis的整合包使用

6.1 导入依赖(在前边的基础上加上这个依赖)：

6.2 java 代码

6.3 测试结果

所有评论(0)

温馨提示：您尚未绑定手机号

你家宝宝