數(shù)據(jù)源可以通過StreamExecutionEnvironment.addSource(sourceFunction)方式來創(chuàng)建，Flink也提供了一些內(nèi)置的數(shù)據(jù)源方便使用，例如readTextFile(path) readFile()，當(dāng)然，也可以寫一個(gè)自定義的數(shù)據(jù)源（可以通過實(shí)現(xiàn)SourceFunction方法，但是無法并行執(zhí)行。或者實(shí)現(xiàn)可以并行實(shí)現(xiàn)的接口ParallelSourceFunction或者繼承RichParallelSourceFunction）

入門

首先做一個(gè)簡(jiǎn)單入門，建立一個(gè)DataStreamSourceApp

Scala

object DataStreamSourceApp {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentsocketFunction(env)env.execute("DataStreamSourceApp")}def socketFunction(env: StreamExecutionEnvironment): Unit = {val data=env.socketTextStream("192.168.152.45", 9999)data.print()} }

這個(gè)方法將會(huì)從socket中讀取數(shù)據(jù)，因此我們需要在192.168.152.45中開啟服務(wù)：

nc -lk 9999

然后運(yùn)行DataStreamSourceApp，在服務(wù)器上輸入：

iie4bu@swarm-manager:~$ nc -lk 9999 apache flink spark

在控制臺(tái)中也會(huì)輸出：

3> apache 4> flink 1> spark

前面的 341表示的是并行度。可以通過設(shè)置setParallelism來操作：

data.print().setParallelism(1)

Java

public class JavaDataStreamSourceApp {public static void main(String[] args) throws Exception {StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();socketFunction(environment);environment.execute("JavaDataStreamSourceApp");}public static void socketFunction(StreamExecutionEnvironment executionEnvironment){DataStreamSource<String> data = executionEnvironment.socketTextStream("192.168.152.45", 9999);data.print().setParallelism(1);} }

自定義添加數(shù)據(jù)源方式

Scala

實(shí)現(xiàn)SourceFunction接口

這種方式不能并行處理。

新建一個(gè)自定義數(shù)據(jù)源

class CustomNonParallelSourceFunction extends SourceFunction[Long]{var count=1Lvar isRunning = trueoverride def run(ctx: SourceFunction.SourceContext[Long]): Unit = {while (isRunning){ctx.collect(count)count+=1Thread.sleep(1000)}}override def cancel(): Unit = {isRunning = false} }

這個(gè)方法首先定義一個(gè)初始值count=1L，然后執(zhí)行的run方法，方法主要是輸出count，并且執(zhí)行加一操作，當(dāng)執(zhí)行cancel方法時(shí)結(jié)束。調(diào)用方法如下：

def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironment// socketFunction(env)nonParallelSourceFunction(env)env.execute("DataStreamSourceApp")}def nonParallelSourceFunction(env: StreamExecutionEnvironment): Unit = {val data=env.addSource(new CustomNonParallelSourceFunction())data.print()}

輸出結(jié)果就是控制臺(tái)一直輸出count值。

無法設(shè)置并行度，除非設(shè)置并行度是1.

val data=env.addSource(new CustomNonParallelSourceFunction()).setParallelism(3)

那么控制臺(tái)報(bào)錯(cuò)：

Exception in thread "main" java.lang.IllegalArgumentException: Source: 1 is not a parallel sourceat org.apache.flink.streaming.api.datastream.DataStreamSource.setParallelism(DataStreamSource.java:55)at com.vincent.course05.DataStreamSourceApp$.nonParallelSourceFunction(DataStreamSourceApp.scala:16)at com.vincent.course05.DataStreamSourceApp$.main(DataStreamSourceApp.scala:11)at com.vincent.course05.DataStreamSourceApp.main(DataStreamSourceApp.scala)

繼承ParallelSourceFunction方法

import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, SourceFunction}class CustomParallelSourceFunction extends ParallelSourceFunction[Long]{var isRunning = truevar count = 1Loverride def run(ctx: SourceFunction.SourceContext[Long]): Unit = {while(isRunning){ctx.collect(count)count+=1Thread.sleep(1000)}}override def cancel(): Unit = {isRunning=false} }

方法的功能跟上面是一樣的。main方法如下：

def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironment// socketFunction(env) // nonParallelSourceFunction(env)parallelSourceFunction(env)env.execute("DataStreamSourceApp")}def parallelSourceFunction(env: StreamExecutionEnvironment): Unit = {val data=env.addSource(new CustomParallelSourceFunction()).setParallelism(3)data.print()}

可以設(shè)置并行度3，輸出結(jié)果如下：

2> 1 1> 1 2> 1 2> 2 3> 2 3> 2 3> 3 4> 3 4> 3

繼承RichParallelSourceFunction方法

class CustomRichParallelSourceFunction extends RichParallelSourceFunction[Long] {var isRunning = truevar count = 1Loverride def run(ctx: SourceFunction.SourceContext[Long]): Unit = {while (isRunning) {ctx.collect(count)count += 1Thread.sleep(1000)}}override def cancel(): Unit = {isRunning = false} } def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironment// socketFunction(env)// nonParallelSourceFunction(env) // parallelSourceFunction(env)richParallelSourceFunction(env)env.execute("DataStreamSourceApp")}def richParallelSourceFunction(env: StreamExecutionEnvironment): Unit = {val data = env.addSource(new CustomRichParallelSourceFunction()).setParallelism(3)data.print()}

Java

實(shí)現(xiàn)SourceFunction接口

import org.apache.flink.streaming.api.functions.source.SourceFunction;public class JavaCustomNonParallelSourceFunction implements SourceFunction<Long> {boolean isRunning = true;long count = 1;@Overridepublic void run(SourceFunction.SourceContext ctx) throws Exception {while (isRunning) {ctx.collect(count);count+=1;Thread.sleep(1000);}}@Overridepublic void cancel() {isRunning=false;} } public static void main(String[] args) throws Exception {StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment(); // socketFunction(environment);nonParallelSourceFunction(environment);environment.execute("JavaDataStreamSourceApp");}public static void nonParallelSourceFunction(StreamExecutionEnvironment executionEnvironment){DataStreamSource data = executionEnvironment.addSource(new JavaCustomNonParallelSourceFunction());data.print().setParallelism(1);}

當(dāng)設(shè)置并行度時(shí)：

DataStreamSource data = executionEnvironment.addSource(new JavaCustomNonParallelSourceFunction()).setParallelism(2);

那么報(bào)錯(cuò)異常：

Exception in thread "main" java.lang.IllegalArgumentException: Source: 1 is not a parallel sourceat org.apache.flink.streaming.api.datastream.DataStreamSource.setParallelism(DataStreamSource.java:55)at com.vincent.course05.JavaDataStreamSourceApp.nonParallelSourceFunction(JavaDataStreamSourceApp.java:16)at com.vincent.course05.JavaDataStreamSourceApp.main(JavaDataStreamSourceApp.java:10)

實(shí)現(xiàn)ParallelSourceFunction接口

import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;public class JavaCustomParallelSourceFunction implements ParallelSourceFunction<Long> {boolean isRunning = true;long count = 1;@Overridepublic void run(SourceContext ctx) throws Exception {while (isRunning) {ctx.collect(count);count+=1;Thread.sleep(1000);}}@Overridepublic void cancel() {isRunning=false;} } public static void main(String[] args) throws Exception {StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment(); // socketFunction(environment); // nonParallelSourceFunction(environment);parallelSourceFunction(environment);environment.execute("JavaDataStreamSourceApp");}public static void parallelSourceFunction(StreamExecutionEnvironment executionEnvironment){DataStreamSource data = executionEnvironment.addSource(new JavaCustomParallelSourceFunction()).setParallelism(2);data.print().setParallelism(1);}

可以設(shè)置并行度，輸出結(jié)果：

1 1 2 2 3 3 4 4 5 5

繼承抽象類RichParallelSourceFunction

public class JavaCustomRichParallelSourceFunction extends RichParallelSourceFunction<Long> {boolean isRunning = true;long count = 1;@Overridepublic void run(SourceContext ctx) throws Exception {while (isRunning) {ctx.collect(count);count+=1;Thread.sleep(1000);}}@Overridepublic void cancel() {isRunning=false;} } public static void main(String[] args) throws Exception {StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment(); // socketFunction(environment); // nonParallelSourceFunction(environment); // parallelSourceFunction(environment);richpParallelSourceFunction(environment);environment.execute("JavaDataStreamSourceApp");}public static void richpParallelSourceFunction(StreamExecutionEnvironment executionEnvironment){DataStreamSource data = executionEnvironment.addSource(new JavaCustomRichParallelSourceFunction()).setParallelism(2);data.print().setParallelism(1);}

輸出結(jié)果：

1 1 2 2 3 3 4 4 5 5 6 6

SourceFunction??ParallelSourceFunction??RichParallelSourceFunction類之間的關(guān)系

?

總結(jié)

以上是生活随笔為你收集整理的Apache Flink 零基础入门（十五）Flink DataStream编程（如何自定义DataSource）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。