1，Flink入坑本地代码 kafka 到 Flink 到 redis

叁歲伎倆 2022-04-16 07:14 181阅读 0赞

object WordCountKafka5Seconds {
    
      //创建redis 映射
    
      class RedisExampleMapper extends RedisMapper[(String)] {
    
        override def getCommandDescription: RedisCommandDescription = {
     //"term_user_group_wxgz_test" 是redis的集合
          new RedisCommandDescription(RedisCommand.HSET,"term_user_group_wxgz_test")
        }
    
        override def getKeyFromData(data: (String)): String = data
    
        override def getValueFromData(data: (String)): String = data
      }
    // zk 地址
      val ZOOKEEPER_HOST = "node1:2181,node2:2181,node3:2181"
    //kafka 地址
      val KAFKA_BROKER = "node4:9092,node5:9092,node6:9092,node7:9092,"
    //kafka的组
      val TRANSACTION_GROUP = "flink_group"
    //kafka topic
      val TOPIC = "topic_test"
    
    //输入redis节点IP
      val REDIS_HOST = "172.10.x.xxx"
    
      def main(args: Array[String]): Unit = {
        System.setProperty("HADOOP_USER_NAME","hadoop")
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        //开启本地模式
    //    val env = StreamExecutionEnvironment.createLocalEnvironment(2)
        env.enableCheckpointing(5000)
        env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    
        val kafkaProps = new Properties()
    
        kafkaProps.setProperty("zookeeper.connect", ZOOKEEPER_HOST)
        kafkaProps.setProperty("bootstrap.servers", KAFKA_BROKER)
        kafkaProps.setProperty("group.id", TRANSACTION_GROUP)
    
        val kafkaConsumer = new FlinkKafkaConsumer010[String](TOPIC, new SimpleStringSchema(), kafkaProps)
        kafkaConsumer.setStartFromLatest()
        val stream: DataStream[String] = env.addSource(kafkaConsumer)
       //可以先注释reids sink代码 测试这里的kafka数据是否收到
        stream.print()
       
        //redis配置选项。//我本地没输入密码 是连接不上redis的
        val conf: FlinkJedisPoolConfig = new  
    
          FlinkJedisPoolConfig.Builder().setHost(REDIS_HOST).setPort(6379).setPassword("xxx").build()
       //输出到redis，是 (String, String) 格式。 
        val  rst: DataStream[(String, String)] = stream.map(x=>(x,"aa"))
        rst.print()
       
    
        env.execute("WordCountKafka5Seconds Example")
    
    
      }
    }