Flink流处理中的数据源与输出目标
在Flink中，无论是流处理还是批处理，数据源（Source）的设计都遵循相似的逻辑。以下是常见的四类数据源：
1. 基于集合的数据源（Collection-based Source）
2. 基于文件的数据源（File-based Source）
3. 基于网络套接字的数据源（Socket-based Source）
4. 自定义数据源（Custom Source）
### 集合数据源
以下代码展示了如何从不同类型的集合中创建数据流：
### 文件数据源
通过文件系统或HDFS读取数据的示例如下：
### 网络套接字数据源
基于TCP套接字的实时数据流可以通过以下方式实现：
### 自定义数据源：以Kafka为例
以下是使用Flink连接Kafka并解析消息的完整示例：
以上内容涵盖了Flink中常见的数据源类型及其使用方法。
import org.apache.flink.streaming.api.scala._
import scala.collection.mutable
object CollectionSourceExample {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 使用基本元素创建流
val simpleStream = env.fromElements("flink", "spark")
simpleStream.print()
// 使用数组创建流
val arrayStream = env.fromCollection(Array("flink", "spark"))
arrayStream.print()
// 使用可变缓冲区创建流
val bufferStream = env.fromCollection(mutable.Buffer("flink", "spark"))
bufferStream.print()
// 使用范围生成流
val rangeStream = env.fromElements((1 to 10): _*)
rangeStream.print()
env.execute("Collection Source Example")
}
}
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 读取本地文件
val localFileStream = env.readTextFile("/path/to/local/file.txt")
localFileStream.print()
// 读取HDFS文件
val hdfsFileStream = env.readTextFile("hdfs://namenode:9000/path/to/hdfs/file.txt")
hdfsFileStream.print()
env.execute("File Source Example")
val env = StreamExecutionEnvironment.getExecutionEnvironment
val socketStream = env.socketTextStream("localhost", 9999)
socketStream.print()
env.execute("Socket Source Example")
import java.util.Properties
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
case class LogData(region: String, method: String, contentType: String)
object KafkaSourceExample {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
val kafkaProps = new Properties()
kafkaProps.setProperty("bootstrap.servers", "localhost:9092")
kafkaProps.setProperty("group.id", "test-group")
val kafkaConsumer = new FlinkKafkaConsumer[String]("test-topic",
new SimpleStringSchema(), kafkaProps)
val stream = env.addSource(kafkaConsumer).map { msg =>
val parts = msg.split(",")
LogData(parts(0), parts(1), parts(2))
}
stream.print()
env.execute("Kafka Source Example")
}
}
