Shulou(Shulou.com)06/01 Report--

This article mainly shows you "Flink how to read data sources", the content is easy to understand, clear, hope to help you solve doubts, the following let the editor lead you to study and learn "how to read data sources Flink" this article.

Read private static void radFromCollection (String [] args) throws Exception {/ / convert parameters to object MultipleParameterTool params = MultipleParameterTool.fromArgs (args) from the collection; / / create batch execution environment / / ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment (); / / create flow processing StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment () / / sets the parallelism of each operator, defaults to cup kernels (test environment) env.setParallelism (2); / sets maximum parallelism env.setMaxParallelism (6); / / reads List collectionData = Arrays.asList ("a", "b", "c", "d") from the collection; DataStreamSource dataStreamSource = env.fromCollection (collectionData) / / read / / env.fromElements ("a", "b", "c", "d") from the array; dataStreamSource.print (); / / dataStreamSource.addSink (new PrintSinkFunction ()); env.execute ();} read StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment () from the file DataStreamSource dataStreamSource = env.readTextFile ("E:\ GIT\ flink-learn\\ flink1\ word.txt", "utf-8"); dataStreamSource.print (); env.execute (); read StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment () from kafka; Properties properties = new Properties (); properties.put ("bootstrap.servers", "10.1.5.130 flink-learn 9092") Properties.put ("zookeeper.connect", "10.2.5.135 my-flink"); 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 kafkaConsumer010 = new FlinkKafkaConsumer010 ("flink", / / topic new SimpleStringSchema (), properties); DataStreamSource dataStreamSource = env.addSource (kafkaConsumer010); dataStreamSource.print (); env.execute (); read from custom Source

Implement org.apache.flink.streaming.api.functions.source.SourceFunction

Public static final class MyDataSource implements SourceFunction {private Boolean running = true; @ Override public void run (SourceContext 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 dataStreamSource = env.addSource (new MyDataSource ()); dataStreamSource.print (); env.execute (); above is all the content of the article "how Flink reads data sources". Thank you for reading! I believe we all have a certain understanding, hope to share the content to help you, if you want to learn more knowledge, welcome to follow the industry information channel!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.