Flink读写kafka主题，并进行数据清洗

Giao.芥末77

833人浏览 · 2021-04-11 14:06:13

Giao.芥末77 · 2021-04-11 14:06:13 发布

半截入土

flink参考
 flink参考

Datastream程序架构

datastream是flink提供给用户使用的用于进行流计算和批处理的api，是对底层流式计算模型的api封装，便于用户编程
一般流程为：

获得一个执行环境；Executiion Environment
加载/创建初始数据；Source
指定转换数据；Transformation
指定存放计算结果的位置；Sink
触发程序执行；流失计算必须的操作，批处理则不必

maven

<flink.version>1.7.2</flink.version>
<kafka.version>2.0.0</kafka.version>

<dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-core -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-shaded-hadoop-2-uber -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-shaded-hadoop-2-uber</artifactId>
            <version>2.4.1-9.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka-0.10 -->
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka-0.11 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>${kafka.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.62</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.11</artifactId>
            <version>${kafka.version}</version>
        </dependency>

简单流水写法

读：设置kafka消费者为flink数据源
写：设置kafka的生产者接受结果

object FlinkReadWriteKafka {
  def main(args: Array[String]): Unit = {
    //首先获取flink流计算环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //kafka props
    val prop = new Properties()
    //指定kafka的Broker地址
    prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "single:9092")
    //指定组的ID
    prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "md")
    //k v 序列化
    prop.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    prop.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    //如果没有记录偏移量，第一次开始消费
    prop.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")


    val ds = env.addSource(
      new FlinkKafkaConsumer[String](
        "event_attendees",
        //对于字符串序列化和反序列化的schema
        new SimpleStringSchema(),
        prop
      ).setStartFromEarliest()//重置游标
    )

    //transform操作
    val dataStream = ds.map(x => {
      val info = x.split(",", -1)
      Array(
        (info(0), info(1).split(" "), "yes"),
        (info(0), info(2).split(" "), "maybe"),
        (info(0), info(3).split(" "), "invited"),
        (info(0), info(4).split(" "), "no")
      )
    }).flatMap(x => x).flatMap(x => x._2.map(y => (x._1, y, x._3))).filter(_._2 != "")
      .map(_.productIterator.mkString(","))

    val prop2 = new Properties()
    prop2.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "single:9092")
    prop2.setProperty(ProducerConfig.RETRIES_CONFIG, "0")
    prop2.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    prop2.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

    dataStream.addSink(new FlinkKafkaProducer[String](
      "single:9092",
      "event_attendees_ff",
      new SimpleStringSchema()))
    //启动流式计算
    env.execute("event_attendees_xf")
  }
}

优化后oop写法

抽象接口–读、写、数据处理

trait Read[T] {
  def read(prop:Properties,tableName:String):DataStream[T]
}

trait Write[T] {
  def write(localhost:String,tableName:String,dataStream:DataStream[T]):Unit
}

trait Transform[T,V] {
  def transform(in:DataStream[T]):DataStream[V]
}

添加读取和写入的数据源

class KafkaSource(env: StreamExecutionEnvironment) extends Read[String] {
  override def read(prop: Properties, tableName: String): DataStream[String] = {
    env.addSource(
      new FlinkKafkaConsumer[String](
        tableName,
        new SimpleStringSchema(),
        prop
      )
    )
  }
}

object KafkaSource {
  def apply(env: StreamExecutionEnvironment): KafkaSource = new KafkaSource(env)
}

class KafkaSink extends Write[String] {
  override def write(localhost: String, tableName: String, dataStream: DataStream[String]): Unit = {
    dataStream.addSink(new FlinkKafkaProducer[String](
      localhost,
      tableName,
      new SimpleStringSchema()
    ))
  }
}

object KafkaSink {
  def apply[T](): KafkaSink = new KafkaSink()
}

依据业务实现数据处理的特质

trait FlikTransform extends Transform[String, String] {
  override def transform(in: DataStream[String]): DataStream[String] = {
    in.map(x => {
      val info = x.split(",", -1)
      Array(
        (info(0), info(1).split(" "), "yes"),
        (info(0), info(2).split(" "), "maybe"),
        (info(0), info(3).split(" "), "invited"),
        (info(0), info(4).split(" "), "no")
      )
    }).flatMap(x => x).flatMap(x => x._2.map(y => (x._1, y, x._3))).filter(_._2 != "")
      .map(_.productIterator.mkString(","))
  }
}

执行器，混入特质

class KTExcutor(readConf: Properties, writelocalhost: String) {
  tran: FlikTransform =>
  def worker(intopic: String, outputtopic: String) = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val kr = new KafkaSource(env).read(readConf, intopic)
    val ds = tran.transform(kr)
    KafkaSink().write(writelocalhost, outputtopic, ds)
    env.execute()
  }
}

动态混入方法，用户执行

object EAtest {
  def main(args: Array[String]): Unit = {
    val prop = new Properties()
    prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "single:9092")
    prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "md")
    prop.setProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "1000")
    prop.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
    prop.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
    prop.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
    val localhost = "single:9092"
    (new KTExcutor(prop, localhost) with FlikTransform)
      .worker("event_attendees", "attendees_AA")
  }
}

天工造物局

北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台

更多推荐

Handy 离线语音转写教程：免费开源，本地跑不连网

你肯定遇到过：语音输入 5 分钟，云转写服务要登录、要付费；或者出差没网，想口述一段会议纪要，发现全都没法用。Handy 离线语音转写工具就是为解决这个问题做的，声音全程不出你的电脑。## 🎯 这个工具是什么Handy 是一款免费、开源、跨平台（Windows / macOS / Linux）的离线语音转写应用。按住快捷键说话，松手后文字直接粘贴进任意输入框，全程不联网。核心能力就三条：

天工造物局

Supabase Edge Functions 编写与部署完整指南：Deno 运行时上的无服务器利器

Supabase Edge Functions 是基于 Deno 运行时编写、面向 Web 的无服务器函数服务，帮你把后端逻辑部署到全球边缘节点。作为 Postgres 开发平台的核心能力之一，它让你用 TypeScript 就能快速搭建低延迟的 API 端点、Webhook 接收器和 AI 集成服务，无需管理任何服务器。本文将从零带你完成一个 Edge Function 的创建、本地调试与全球部

天工造物局

celld从v0.1.0到v0.2.0为何不能滚动升级：两个破坏性变更详解

celld 是一个自托管、分布式的 Durable Objects 开源守护进程，它在你自己的机器上运行 Cloudflare Workers，每个对象对应一个独立的 SQLite 数据库，并持续复制到你自己拥有的 S3 兼容存储桶中。如果你正准备把 v0.1.0 的集群（fleet）升级到 v0.2.0，这篇文章必须先读：这次升级**不能滚动升级**，必须采用"先停光、再启动"的一次性发布方式。