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

This article mainly explains "how to read and write txt files with Flink Connectors". Interested friends may wish to have a look. The method introduced in this paper is simple, fast and practical. Let's let the editor take you to learn how to read and write txt files with Flink Connectors.

Open the txt file by using the Flink DataSet Connectors data flow connector and provide data stream input and output operations

Sample environment

Java.version: 1.8.xflink.version: 1.11.1

Sample data source (project code cloud download)

Building Development Environment and data of Flink system example

Sample module (pom.xml)

DataStream Connectors and sample Module of Flink system

Data stream input

TextSource.java

Package com.flink.examples.file;import org.apache.commons.lang3.StringUtils;import org.apache.commons.lang3.time.DateUtils;import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.java.DataSet;import org.apache.flink.api.java.ExecutionEnvironment;import org.apache.flink.util.Collector;import scala.Tuple7;import java.util.Date / * * @ Description reads the contents from the txt file and outputs them to DataSet * / public class TextSource {public static void main (String [] args) throws Exception {ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment (); String filePath = "D:\\ Workspaces\\ idea_2\\ flink-examples\ connectors\\ src\\ main\\ resources\\ user.txt" DataSet dataSet = env.readTextFile (filePath) .flatMap (new FlatMapFunction () {@ Override public void flatMap (String value, Collector out) throws Exception {if (StringUtils.isNotBlank (value)) {String [] values = value.split (",") Out.collect (Tuple7.apply (Integer.parseInt (values [0]), values [1], Integer.parseInt (values [2]), Integer.parseInt (values [3])) Values [4], DateUtils.parseDate (values [5], "yyyy-MM-dd HH:mm:ss"), Long.parseLong (values [6])) }); dataSet.print ();}}

Data stream output

TextSink.java

Package com.flink.examples.file;import org.apache.flink.api.java.DataSet;import org.apache.flink.api.java.ExecutionEnvironment;import org.apache.flink.core.fs.FileSystem;import org.apache.flink.types.Row;/** * @ Description writes DataSet data to the txt file * / public class TextSink {public static void main (String [] args) throws Exception {ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment () / / first create the file String filePath = "D:\ Workspaces\\ idea_2\\ flink-examples\\ connectors\\ src\\ main\\ resources\\ user.txt"; / / convert the entity to a Row object, new Row (number of fields) Row row = Row.of (15, "chen1", 40, 1, "CN", "2020-09-08 00:00:00", 1599494400000L) / / convert to dataSet DataSet dataSet = env.fromElements (row); / / write the content to File. If the file already exists, it will be overwritten by dataSet.writeAsText (filePath, FileSystem.WriteMode.OVERWRITE) .setParallelism (1); env.execute ("fline file sink");}}

Data presentation

At this point, I believe you have a deeper understanding of "how to read and write txt files with Flink Connectors". You might as well do it in practice. Here is the website, more related content can enter the relevant channels to inquire, follow us, continue to learn!

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.