Flink Learning 12: DataStreaming API

1. Concept

2.DatStream programming

There are 3 parts that need to be implemented by code, 1. Data source 2. Conversion 3. Sink specified output format

2.1 Create environment

val env = ExecutionEnvironment.getExecutionEnvironment

The statement is relatively fixed

2.2 Data Source

There are three main built-in data sources:

2.2.1 Text Data Source

import org.apache.flink.streaming.api.scala._import org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentobject readText {def main(args: Array[String]): Unit = {//create envval env = StreamExecutionEnvironment.getExecutionEnvironment//create datasourceval dataStream = env.readTextFile(filePath = "C:\\doc\\temp\\1. Introduction to the main line item.txt")//printdataStream.print()//executeenv.execute()}}

2.2.2 Socket Streams

Use the NC that comes with the Ubantu system to generate a socket data source

Then write code to monitor socket data

import org.apache.flink.api.scala._import org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentobject socketSourceTest {def main(args: Array[String]): Unit = {//create envval env = StreamExecutionEnvironment.getExecutionEnvironment//create datasourceval socketDataStream = env.socketTextStream("localhost", 9999, '\n')//printsocketDataStream.print()//executeenv.execute()}}

Output results

2.2.3 Collection Data Sources

import org.apache.flink.api.scala._object CollectionSourceTest {def main(args: Array[String]): Unit = {//create envval env = ExecutionEnvironment.getExecutionEnvironment//collection data sourceval collectionDataStream = env.fromElements(Tuple1(1L, 2L), Tuple1(3L, 4L))//printcollectionDataStream.print()}}Output: