Flink 实时将数据写到 Redis

亦凉 2023-02-10 05:16 22阅读 0赞

1.  ###### RedisSinkDemo 代码 ######
    
        import java.net.InetSocketAddress
        import java.util
        
        import org.apache.flink.streaming.api.scala._
        import org.apache.flink.streaming.connectors.redis.RedisSink
        import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisClusterConfig
        import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}
        
        
        object RedisSinkDemo {
        
          def main(args: Array[String]): Unit = {
        
            val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
        
            val socketHost: String = "node01"
            val socketPort: Int = 7777
            val redisHost: String = "node02"
            val redisPort: Int = 7777
        
            val socketDs: DataStream[String] = streamEnv.socketTextStream(socketHost, socketPort)
        
            val wordCount: DataStream[(String, Int)] =
              socketDs.flatMap(_.split(" ")).map((_, 1)).keyBy(0).sum(1)
        
            //连接redis的配置
            val nodes = new util.HashSet[InetSocketAddress](
              util.Arrays.asList(
                new InetSocketAddress(redisHost, redisPort)
              ))
            val config: FlinkJedisClusterConfig = new FlinkJedisClusterConfig.Builder()
              .setNodes(nodes).build()
            //写入redis
            wordCount.addSink(new RedisSink[(String, Int)](config, new MyRedisMapper))
            streamEnv.execute("Redis Sink Demo")
          }
        }
        
        class MyRedisMapper extends RedisMapper[(String, Int)] {
          override def getCommandDescription: RedisCommandDescription = {
            new RedisCommandDescription(RedisCommand.HSET, "wordcount")
          }
        
          override def getKeyFromData(t: (String, Int)): String = {
            t._1
          }
        
          override def getValueFromData(t: (String, Int)): String = {
            t._2.toString
          }
        }
2.  ###### 自行编译打包 flink-connector-redis\_2.11 ######
    
    下载地址: https://bahir.apache.org/downloads/flink/  
    github: https://github.com/apache/bahir-flink.git
    
    编译命令：
    
        mvn -T8C -DskipTests=true clean install
    
    本地仓库安装命令:
    
        mvn install:install-file --settings=D:\development\apache-maven-3.3.9\conf\settings-local.xml -Dfile=D:\flink-connector-redis_2.11-1.1-SNAPSHOT.jar -DgroupId=org.apache.bahir -DartifactId=flink-connector-redis_2.11 -Dversion=1.1-SNAPSHOT -Dpackaging=jar
    
    **遇到的问题:**  
    使用 RedisCommand.HSET 会在 flink 程序重新启动时会重置 redis 中存储的 key，但由于目前该版本不支持 hincriby 命令，故自行添加支持该命令。代码请查看 https://github.com/Git-Charles/bahir-flink.git
    
    使用 RedisCommand.HINCRBY 时，不能加 sum(1) 函数，这里 flink 做了累加。  
    上述代码修改为:
    
        ...
        val wordCount: DataStream[(String, Int)] =
              socketDs.flatMap(_.split(" ")).map((_, 1)).keyBy(0)
        ...
        new RedisCommandDescription(RedisCommand.HINCRBY, "wordcount")
        ...
3.  ###### pom.xml 文件部分内容 ######
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
            <scala.version>2.11.12</scala.version>
            <scala.version.large>2.11</scala.version.large>
            
            <flink.version>1.9.2</flink.version>
            <jedis.version>2.9.0</jedis.version>
            <flink.connector.redis.version>1.1-SNAPSHOT</flink.connector.redis.version>
            <scope.level>compile</scope.level>
        </properties>
        
        <repositories>
            
            <repository>
                <id>cloudera</id>
                <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
                <name>Cloudera Repositories</name>
                <releases>
                    <enabled>true</enabled>
                </releases>
                <snapshots>
                    <enabled>false</enabled>
                </snapshots>
            </repository>
        </repositories>
        
        <dependencies>
            <dependency>
                <groupId>redis.clients</groupId>
                <artifactId>jedis</artifactId>
                <version>${jedis.version}</version>
            </dependency>
            
            <dependency>
                <groupId>org.scala-lang</groupId>
                <artifactId>scala-library</artifactId>
                <version>${scala.version}</version>
                
            </dependency>
            <dependency>
                <groupId>org.apache.bahir</groupId>
                <artifactId>flink-connector-redis_${scala.version.large}</artifactId>
                <version>${flink.connector.redis.version}</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.apache.flink</groupId>
                        <artifactId>flink-streaming-java_2.11</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-scala_${scala.version.large}</artifactId>
                <version>${flink.version}</version>
                <scope>${scope.level}</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>${scope.level}</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-scala_${scala.version.large}</artifactId>
                <version>${flink.version}</version>
                <scope>${scope.level}</scope>
            </dependency>
        </dependencies>