flink 1.17 sink (输入) and source (输出) Scala

Sources 输出#

Flink Local File System#

example:

1
import org.apache.flink.api.common.serialization.SimpleStringEncoder
2
import org.apache.flink.configuration.MemorySize
3
import org.apache.flink.connector.file.sink.FileSink
4
import org.apache.flink.core.fs.Path
5
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy
6
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
7

8

9
object Main {
10
  def main(args: Array[String]): Unit = {
11
    val env = StreamExecutionEnvironment.getExecutionEnvironment
12

13
    val value = env.readTextFile("/source/1.txt")
14

15
    value.print()
16

17
    value.sinkTo(sink)
18

19
    env.execute()
20
  }
21
}

Flink HDFS File System#

example:

1
import org.apache.flink.api.common.serialization.SimpleStringEncoder
2
import org.apache.flink.configuration.MemorySize
3
import org.apache.flink.connector.file.sink.FileSink
4
import org.apache.flink.core.fs.Path
5
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy
6
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
7

8

9
object Main {
10
  def main(args: Array[String]): Unit = {
11
    val env = StreamExecutionEnvironment.getExecutionEnvironment
12

13
    val value = env.readTextFile("hdfs://xxx.xxx.xxx.xxx:9000/data/word.txt")
14

15
    value.print()
16

17
    value.sinkTo(sink)
18

19
    env.execute()
20
  }
21
}

Flink KAFKA Source#

1
package org.example
2

3
import org.apache.flink.api.common.eventtime.WatermarkStrategy
4
import org.apache.flink.api.common.serialization.SimpleStringSchema
5
import org.apache.flink.connector.kafka.source.KafkaSource
6
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
7
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
8

9

10
object Main {
11
  def main(args: Array[String]): Unit = {
12

13
    val env = StreamExecutionEnvironment.getExecutionEnvironment
14

15
    val value = KafkaSource
16
      .builder()
17
      .setBootstrapServers("172.18.38.32:9092,172.18.38.33:9092,172.18.38.34:9092") // setting kafka bootstrap servers
18
      .setTopics("test") // setting kafka topic
19
      .setGroupId("aasds") // setting kafka group_id
20
      .setStartingOffsets(OffsetsInitializer.earliest()) // setting kafka offset
21
      .setValueOnlyDeserializer(new SimpleStringSchema()) // value deserializer
22
      .build()
23

24
    val value1 = env.fromSource(
25
      value,
26
      WatermarkStrategy.noWatermarks(),
27
      "Demo Source"
28
    )
29

30
    value1.print()
31

32
    env.execute()
33
  }
34
}

Sinks 输出#

Flink Local File System#

使用本地系统输出

pom.xml

1
        <dependency>
2
            <groupId>org.apache.flink</groupId>
3
            <artifactId>flink-connector-files</artifactId>
4
            <version>${flink.version}</version>
5
        </dependency>

example:

1
import org.apache.flink.api.common.serialization.SimpleStringEncoder
2
import org.apache.flink.configuration.MemorySize
3
import org.apache.flink.connector.file.sink.FileSink
4
import org.apache.flink.core.fs.Path
5
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy
6
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
7

8

9
object Main {
10
  def main(args: Array[String]): Unit = {
11
    val env = StreamExecutionEnvironment.getExecutionEnvironment
12

13
    val value = env.readTextFile("hdfs://xxx.xxx.xxx.xxx:9000/data/word.txt")
14

15
    val sink: FileSink[String] = FileSink
16
      .forRowFormat(new Path("./output"), new SimpleStringEncoder[String]("UTF-8"))
17
      .withRollingPolicy(
18
        DefaultRollingPolicy.builder()
19
          .withMaxPartSize(MemorySize.ofMebiBytes(1024))
20
          .build())
21
      .build()
22

23
    value.print()
24

25
    value.sinkTo(sink)
26

27
    env.execute()
28
  }
29
}

Flink HDFS File System#

pom.xml

1
        <dependency>
2
            <groupId>org.apache.flink</groupId>
3
            <artifactId>flink-connector-files</artifactId>
4
            <version>${flink.version}</version>
5
        </dependency>

使用hadoop系统输出

example:

1
import org.apache.flink.api.common.serialization.SimpleStringEncoder
2
import org.apache.flink.configuration.MemorySize
3
import org.apache.flink.connector.file.sink.FileSink
4
import org.apache.flink.core.fs.Path
5
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy
6
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
7

8

9
object Main {
10
  def main(args: Array[String]): Unit = {
11
    val env = StreamExecutionEnvironment.getExecutionEnvironment
12

13
    val value = env.readTextFile("hdfs://xxx.xxx.xxx.xxx:9000/data/word.txt")
14

15
    val sink: FileSink[String] = FileSink
16
      .forRowFormat(new Path("hdfs://xxx.xxx.xxx.xxx:9000/output/flink"), new SimpleStringEncoder[String]("UTF-8"))
17
      .withRollingPolicy(
18
        DefaultRollingPolicy.builder()
19
          .withMaxPartSize(MemorySize.ofMebiBytes(1024))
20
          .build())
21
      .build()
22

23
    value.print()
24

25
    value.sinkTo(sink)
26

27
    env.execute()
28
  }
29
}