您好,登錄后才能下訂單哦!
這篇文章主要為大家展示了“Flink如何讀取數(shù)據(jù)源”,內(nèi)容簡(jiǎn)而易懂,條理清晰,希望能夠幫助大家解決疑惑,下面讓小編帶領(lǐng)大家一起研究并學(xué)習(xí)一下“Flink如何讀取數(shù)據(jù)源”這篇文章吧。
private static void radFromCollection(String[] args) throws Exception { //將參數(shù)轉(zhuǎn)成對(duì)象 MultipleParameterTool params = MultipleParameterTool.fromArgs(args); //創(chuàng)建批處理執(zhí)行環(huán)境 // ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); //創(chuàng)建流程處理 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //設(shè)置每個(gè)算子的的并行度,默認(rèn)為cup核數(shù)(測(cè)試環(huán)境下) env.setParallelism(2); //設(shè)置最大并行度 env.setMaxParallelism(6); //從集合中讀取 List<String> collectionData = Arrays.asList("a", "b", "c", "d"); DataStreamSource<String> dataStreamSource = env.fromCollection(collectionData); //從數(shù)組中讀取 // env.fromElements("a", "b", "c", "d"); dataStreamSource.print(); //dataStreamSource.addSink(new PrintSinkFunction<>()); env.execute(); }
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource<String> dataStreamSource = env.readTextFile("E:\\GIT\\flink-learn\\flink1\\word.txt", "utf-8"); dataStreamSource.print(); env.execute();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); Properties properties = new Properties(); properties.put("bootstrap.servers", "10.1.5.130:9092"); properties.put("zookeeper.connect", "10.2.5.135:2181"); properties.put("group.id", "my-flink"); properties.put("auto.offset.reset", "latest"); properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); FlinkKafkaConsumer010<String> kafkaConsumer010 = new FlinkKafkaConsumer010<>( "flink",// topic new SimpleStringSchema(), properties ); DataStreamSource<String> dataStreamSource = env.addSource(kafkaConsumer010); dataStreamSource.print(); env.execute();
實(shí)現(xiàn) org.apache.flink.streaming.api.functions.source.SourceFunction
public static final class MyDataSource implements SourceFunction<String> { private Boolean running = true; @Override public void run(SourceContext<String> sourceContext) throws Exception { Random random = new Random(); while (running) { double data = random.nextDouble() * 100; sourceContext.collectWithTimestamp(String.valueOf(data), System.currentTimeMillis()); TimeUnit.SECONDS.sleep(1); } } @Override public void cancel() { this.running = false; } }
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource<String> dataStreamSource = env.addSource(new MyDataSource()); dataStreamSource.print(); env.execute();
以上是“Flink如何讀取數(shù)據(jù)源”這篇文章的所有內(nèi)容,感謝各位的閱讀!相信大家都有了一定的了解,希望分享的內(nèi)容對(duì)大家有所幫助,如果還想學(xué)習(xí)更多知識(shí),歡迎關(guān)注億速云行業(yè)資訊頻道!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。