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

How to enter the Flinksink in ApacheFlink, I believe that many inexperienced people are at a loss about it. Therefore, this paper summarizes the causes and solutions of the problem. Through this article, I hope you can solve this problem.

Where to Sink the data in DataSet. Use the corresponding OutPutFormat, or you can use a custom sink, which may be written into hbase or hdfs.

WriteAsText () / TextOutputFormat, written as String

WriteAsCsv (...) / CsvOutputFormat, written in CSV

Print () / printToErr () / print (String msg) / printToErr (String msg) as standard output

WriteAsTextobject DataSetSinkApp {def main (args: Array [String]): Unit = {val environment = ExecutionEnvironment.getExecutionEnvironment val data = 1.to (10) val text = environment.fromCollection (data) val filePath = "E:/test" text.writeAsText (filePath) environment.execute ("DataSetSinkApp")}}

If the E:/test file or folder exists, the execution will not succeed. Unless a WriteMode.OVERWRITE is added

Text.writeAsText (filePath, WriteMode.OVERWRITE)

This creates a new test file under disk E, with contents ranging from 1 to 10.

So how do I save it to a folder?

Text.writeAsText (filePath, WriteMode.OVERWRITE) .setParallelism (2)

Set the parallelism to 2, so it is saved to the test folder, two files 1 and 2

By default, if parallelism is not set, the result is written to a file, and if parallelism is set, each parallelism corresponds to an output.

Java public static void main (String [] args) throws Exception {ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment (); List info = new ArrayList (); for (int I = 1bot I)

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.