Flink_DataStream 的 Transformation

Flink_DataStream 的 Transformation"/>

Flink_DataStream 的 Transformation

文章目录

• DataStream 的 Transformation
• • 1. KeyBy
  • 2. Connect
  • 3. Split 和 select
• 数据输出 Data Sinks
• • 1.sink 到 kafka
  • 2. sink 到 mysql

DataStream 的 Transformation

1. KeyBy

逻辑上将一个流分成不相交的分区，每个分区包含相同键的元素。在内部，这是通过散 列分区来实现的

package com.czxy.flink.stream.transformationimport org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}//keyBy分组操作算子
object StreamKeyBy {def main(args: Array[String]): Unit = {//1.创建执行环境val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//2.构建数据集import org.apache.flink.api.scala._val elementSource: DataStream[String] = env.fromElements("hadoop hadoop spark hive flink flink")//3.数据组合成元祖类型val wordAndOne: DataStream[(String, Int)] = elementSource.flatMap(x=>x.split(" ")).map((_,1))//4.进行分组val KeyedStream: KeyedStream[(String, Int), Tuple] = wordAndOne.keyBy(0)//5.聚合计算val result: DataStream[(String, Int)] = KeyedStream.reduce((v1,v2)=>(v1._1,v1._2+v2._2))//6.打印输出result.print().setParallelism(1)//7.执行程序env.execute("StreamKeyBy")}
}

2. Connect

用来将两个 dataStream 组装成一个 ConnectedStreams 而且这个 connectedStream 的组成结构就是保留原有的 dataStream 的结构体；这样我们 就可以把不同的数据组装成同一个结构

代码示例：

package com.czxy.flink.stream.transformationimport org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.{ConnectedStreams, DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
/*** Connect合并流 */
object StreamConnect {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironmentval source1: DataStream[Long] = env.addSource(new NoParallelSource()).setParallelism(1)val source2: DataStream[Long] = env.addSource(new NoParallelSource()).setParallelism(1)val connectedStreams: ConnectedStreams[Long, Long] = source1.connect(source2)val result: DataStream[String] = connectedStreams.map(item1 => {"item1: " + item1},item2 => {"item2: " + item2})result.print()env.execute("StreamConnect")}//实现一个单线程的，数据从1开始递增的数据集class NoParallelSource extends SourceFunction[Long]() {var number: Long = 1Lvar isRunning: Boolean = trueoverride def run(ctx: SourceFunction.SourceContext[Long]): Unit = {while (isRunning) {ctx.collect(number)number += 1Thread.sleep(1)if (number > 5) {cancel()}}}override def cancel(): Unit = {isRunning = false}}
}

3. Split 和 select

代码实现

package com.czxy.flink.stream.transformationimport org.apache.flink.streaming.api.scala.{DataStream, SplitStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
/*** 需求：* 给出数据 1, 2, 3, 4, 5, 6, 7* 请使用 split 和 select 把数据中的奇偶数分开， 并打印出奇数*/
object StreamSplit {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironmentval source: DataStream[Int] = env.fromElements(1, 2, 3, 4, 5, 6, 7)val splitStream: SplitStream[Int] = source.split(x => {(x % 2) match {case 0 => List("偶数")case 1 => List("奇数")}})val result: DataStream[Int] = splitStream.select("奇数")result.print()env.execute("StreamSplit")}
}

数据输出 Data Sinks

• 3 、将数据 sink 到本地文件(参考批处理)
• 4 、Sink 到本地集合(参考批处理)
• 5 、Sink 到 HDFS(参考批处理)

1.sink 到 kafka

代码示例

package com.czxy.flink.stream.sinkimport java.util.Propertiesimport org.apache.flink.streaming.api.datastream.DataStreamSink
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011
import org.apache.flink.streaming.util.serialization.SimpleStringSchemaobject StreamKafkaSink {def main(args: Array[String]): Unit = {//1.创建执行环境val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//2.构建数据集import org.apache.flink.api.scala._val source: DataStream[String] = env.fromElements("1,小丽,北京,女")//3.设置kafka的配置信息val topic="test"val properties: Properties = new Properties()properties.setProperty("bootstrap.servers","node01:9092")val flinkKafkaProducer: FlinkKafkaProducer011[String] = new FlinkKafkaProducer011[String](topic,new SimpleStringSchema(),properties)val result: DataStreamSink[String] = source.addSink(flinkKafkaProducer)//    source.addSink(flinkKafkaProducer)
//    source.print()env.execute("StreamKafkaSink")}
}

2. sink 到 mysql

package com.czxy.flink.stream.sinkimport java.sql.{Connection, DriverManager, PreparedStatement}import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.datastream.DataStreamSink
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._object StreamMysqlSink {case class Student(stuId: Int, stuName: String, stuAddr: String, stuSex: String)def main(args: Array[String]): Unit = {//1.创建执行环境val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//2.准备数据val source: DataStream[Student] = env.fromElements(//Student(8, "小青", "广州", "女"),Student(9, "wangman", "beijing", "nv"))val result: DataStreamSink[Student] = source.addSink(new MysqlSink())env.execute("StreamMysqlSink")}class MysqlSink extends RichSinkFunction[Student]() {var connection: Connection = nullvar ps: PreparedStatement = nulloverride def open(parameters: Configuration): Unit = {val driver = "com.mysql.jdbc.Driver"val url = "jdbc:mysql://localhost:3306/test?characterEncoding=utf-8&useSSL=false"val username = "root"val password = "root"//1:加载驱动Class.forName(driver)//2：创建连接connection = DriverManager.getConnection(url, username, password)val sql ="""|insert into student(id , name , addr , sex)values(?,?,?,?);|""".stripMargin//3:获得执行语句ps = connection.prepareStatement(sql)}
//    //关闭连接操作
//    override def close(): Unit = {
//      if (connection != null) {
//        connection.close() 
//      }
//      if (ps != null) {
//        ps.close()
//      }
//    }//每个元素的插入，都要触发一次 invoke，这里主要进行 invoke 插入 每条数据执行一次override def invoke(value: Student): Unit = {try{//4.组装数据，执行插入操作ps.setInt(1, value.stuId)ps.setString(2, value.stuName)ps.setString(3, value.stuAddr)ps.setString(4, value.stuSex)ps.executeUpdate()}catch{case e:Exception=>println(e.getMessage)}}}
}