Flink 의 자바 8

주소:https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/java8.html
자바 8 은 새로운 기능 을 도입 하여 더욱 빠 르 고 선명 한 프로 그래 밍 을 할 수 있 습 니 다. 가장 중요 한 특성 은 람 다 표현 식 입 니 다. 자바 함수 식 프로 그래 밍 의 문 을 열 었 습 니 다.Lambda 표현 식 은 익명 함 수 를 실현 하고 전달 할 수 있 습 니 다!
예 를 들 면:
words. map {x = > (x, 1)} 의 x = > (x, 1) 는 단일 변수 에서 이원 그룹 까지 의 익명 함수 입 니 다!
최신 Flink 버 전 은 자바 의 Lambda 표현 식 프로 그래 밍 을 지원 합 니 다. 이 문 서 는 Lambda 표현 식 프로 그래 밍 과 현재 의 한 계 를 설명 할 것 입 니 다!프로 그래 밍 가이드
Examples
다음 예 는 Lambda 표현 식 의 프로그램 입 니 다. map 함 수 는 원 수 를 제곱 한 후에 인쇄 합 니 다. 그 중의 map 안의 함 수 는 익명 함수 입 니 다. 함수 유형 을 설명 할 필요 가 없습니다. 자바 8 은 유형 을 추정 할 수 있 습 니 다.

env.fromElements(1, 2, 3)
// returns the squared i
.map(i -> i*i)
.print();

다음 예 는 Collector 의 종 류 를 추정 할 수 없 기 때문에 설명 형식 이 필요 합 니 다.

DataSet<Integer> input = env.fromElements(1, 2, 3);

// collector type must be declared
input.flatMap((Integer number, Collector<String> out) -> {
    StringBuilder builder = new StringBuilder();
    for(int i = 0; i < number; i++) {
        builder.append("a");
        out.collect(builder.toString());
    }
})
// returns (on separate lines) "a", "a", "aa", "a", "aa", "aaa"
.print();

 
   
       DataSet     ：
  
DataSet<Integer> input = env.fromElements(1, 2, 3);

// collector type must not be declared, it is inferred from the type of the dataset
DataSet<String> manyALetters = input.flatMap((number, out) -> {
    StringBuilder builder = new StringBuilder();
    for(int i = 0; i < number; i++) {
       builder.append("a");
       out.collect(builder.toString());
    }
});
  
다음 코드 는 lambda 표현 식 을 사용 하 는 wordcount 프로그램 을 보 여 줍 니 다.
DataSet<String> input = env.fromElements("Please count", "the words", "but not this");

// filter out strings that contain "not"
input.filter(line -> !line.contains("not"))
// split each line by space
.map(line -> line.split(" "))
// emit a pair  for each array element
.flatMap((String[] wordArray, Collector<Tuple2<String, Integer>> out)
    -> Arrays.stream(wordArray).forEach(t -> out.collect(new Tuple2<>(t, 1)))
    )
// group and sum up
.groupBy(0).sum(1)
// print
.print();
  
다음은 자바 8 사용 제한 제약 등 문제 입 니 다. 구체 적 인 참고 문서 입 니 다!
Compiler Limitations 
Currently, Flink only supports jobs containing Lambda Expressions completely if they are compiled with the Eclipse JDT compiler contained in Eclipse Luna 4.4.2 (and above). 
Only the Eclipse JDT compiler preserves the generic type information necessary to use the entire Lambda Expressions feature type-safely. Other compilers such as the OpenJDK’s and Oracle JDK’s  javac  throw away all generic parameters related to Lambda Expressions. This means that types such as  Tuple2 or Collector  declared as a Lambda function input or output parameter will be pruned to  Tuple2  or  Collector  in the compiled  .class  files, which is too little information for the Flink Compiler. 
How to compile a Flink job that contains Lambda Expressions with the JDT compiler will be covered in the next section. 
However, it is possible to implement functions such as  map()  or  filter()  with Lambda Expressions in Java 8 compilers other than the Eclipse JDT compiler as long as the function has no  Collector s or  Iterable s and only if the function handles unparameterized types such as  Integer ,  Long ,  String ,  MyOwnClass  (types without Generics!). 
Compile Flink jobs with the Eclipse JDT compiler and Maven 
If you are using the Eclipse IDE, you can run and debug your Flink code within the IDE without any problems after some configuration steps. The Eclipse IDE by default compiles its Java sources with the Eclipse JDT compiler. The next section describes how to configure the Eclipse IDE. 
If you are using a different IDE such as IntelliJ IDEA or you want to package your Jar-File with Maven to run your job on a cluster, you need to modify your project’s  pom.xml  file and build your program with Maven. The quickstart contains preconfigured Maven projects which can be used for new projects or as a reference. Uncomment the mentioned lines in your generated quickstart  pom.xml  file if you want to use Java 8 with Lambda Expressions. 
Alternatively, you can manually insert the following lines to your Maven  pom.xml  file. Maven will then use the Eclipse JDT compiler for compilation. 



    
    maven-compiler-plugin
    
        1.8
        1.8
        jdt
    
    
        
        
            org.eclipse.tycho
            tycho-compiler-jdt
            0.21.0
        
    
  
If you are using Eclipse for development, the m2e plugin might complain about the inserted lines above and marks your  pom.xml  as invalid. If so, insert the following lines to your  pom.xml . 



    
        org.apache.maven.plugins
        maven-compiler-plugin
        [3.1,)
        
            testCompile
            compile
        
    
    
        
    
  
Run and debug Flink jobs within the Eclipse IDE 
First of all, make sure you are running a current version of Eclipse IDE (4.4.2 or later). Also make sure that you have a Java 8 Runtime Environment (JRE) installed in Eclipse IDE ( Window  ->  Preferences  ->  Java  ->  Installed JREs ). 
Create/Import your Eclipse project. 
If you are using Maven, you also need to change the Java version in your  pom.xml  for the  maven-compiler-plugin . Otherwise right click the  JRE System Library  section of your project and open the  Properties  window in order to switch to a Java 8 JRE (or above) that supports Lambda Expressions. 
The Eclipse JDT compiler needs a special compiler flag in order to store type information in  .class  files. Open the JDT configuration file at  {project directoy}/.settings/org.eclipse.jdt.core.prefs  with your favorite text editor and add the following line: 
org.eclipse.jdt.core.compiler.codegen.lambda.genericSignature=generate
  
If not already done, also modify the Java versions of the following properties to  1.8  (or above): 
org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
org.eclipse.jdt.core.compiler.compliance=1.8
org.eclipse.jdt.core.compiler.source=1.8
  
After you have saved the file, perform a complete project refresh in Eclipse IDE. 
If you are using Maven, right click your Eclipse project and select  Maven  ->  Update Project... . 
You have configured everything correctly, if the following Flink program runs without exceptions: 
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.fromElements(1, 2, 3).map((in) -> new Tuple1<String>(" " + in)).print();
env.execute();