How to get started with Flinksink in ApacheFlink
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)