Preface

One more Flink Of stream Of example, Say it in advance , The example on the official website is a bit of a pit .

One 、stream Example

Copy to my directory （ By the way, the next good tool is really fragrant ,idea Code that can be copied , Paste automatically create classes with me ）

What do you want to write , That's true. , Don't ask yet , Keep looking down .

Two 、example Arrangement

1. Depend on the introduction of

You copy it to your demo project , When automatic bag drawing , You will find a lot of missing objects .
The first thing we need to do is introduce flink-connector-files

<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-connector-files</artifactId>
	<version>${
    flink.version}</version>
</dependency>

2. Why is there TextLineInputFormat class

Finish the steps above , You should find that this class is missing . First of all , The example here is actually written according to the latest snapshot version , Go back and see example Of pom Will find .

I started from maven Download to this version , use luyten open jar, I found that this class really exists .

So I created such a class , Here's the thing , If you use the latest version 1.14.4 edition , In fact, that line of code can be written in another way .

FileSource.FileSourceBuilder<String> builder =
   // FileSource.forRecordStreamFormat(
   //new TextLineInputFormat(), params.getInputs().get()); 1.16-SNAPSHOT edition 
  FileSource.forRecordStreamFormat(new TextLineFormat()
  , params.getInputs().get());    //1.14.4 edition 

The right of final interpretation , Or from the official website , On the official website 1.14.4 Version of Search Box input Text, Will appear Text Files The drop-down , Click on , You will jump to the syntax usage example

3. There are also error reporting tips

Click on MemorySize Methods ofMebiBytes See the realization , Make sure you can write directly long value , in addition , below Duration It's also popular here , Change ：

counts.sinkTo(
          FileSink.<Tuple2<String, Integer>>forRowFormat(
                  params.getOutput().get(), new SimpleStringEncoder<>())
              .withRollingPolicy(
                  DefaultRollingPolicy.builder()
                      .withMaxPartSize(20)
                      .withRolloverInterval(10)
                      .build())
              .build())
      .name("file-sink");

3、 ... and 、 function example

Report errors , I checked the reasons and didn't explain , Look, it literally means , Cannot use private final bytes ...
PS： Speaking of this, I suddenly thought ,jdk stay 9 Will be realized by char[] Changed to byte[] 了 , The meaning is that it can save memory .
In fact, I doubt whether it is jdk17 Yes bug, Finally, I suddenly thought of building the official website demo Where the project script is located 2 Sentence ：

therefore , Switch demo Of build＆run Of jdk edition ,idea Switch here ：

Run again

The familiar partition statistics are coming out again , Are you very happy ！！！

summary

1、 Read the official website example Also a lot of , Later learning may not be shared with everyone , Some of them are definitely not needed for the time being .
2、 my CSDN Learning members Flink This course has 100 lesson , Learning should be accelerated , There may be no special time to study later , Only in actual combat get 了 .
I hope I can help you , Off work , get home .