Home
Data & Data Engineering
Docker를 사용한 Apache Flink와 Flink Job 올리기(2) - Flink Job Example
울이
울이
December 25, 2021
1 min

Table Of Contents

01
개발환경
02
Flink Job Code
03
Flink Job 등록하기

Flink 작업을 위해서 Apache Flink 책인 <Stream Processing with Apache Flink: Fundamentals, Implementation, and Operation of Streaming Applications> 를 참고했으며 예제코드를 다운받아 작성했다.

사실상 코드는 책에서 제공 해 주는 코드를 실행하기 때문에, 주의 해 주어야 할 것은 Maven 설정을 잘 해주는 것과 Java 환경설정, 그리고 JAR 파일을 잘 말아주는 것 밖에는 할 것이 없었다.

Flink에 조금 더 익숙 해 진다면 직접 Flink Job을 만들어 보고싶은 생각이… -> 내가 아직 잘 사용 못하는 것 같다.. 일반 consumer application이랑 차이를 모르겠다..

개발환경

  • macOS
  • Java 15
  • Intellij
  • Maven

Flink Job Code

소스코드 다운받기

소스코드 출처 - 링크

git clone [https://github.com/streaming-with-flink/examples-java](https://github.com/streaming-with-flink/examples-java)

소스코드 Open하기 - Intellij

  • Intellij → File → Open → [examples-java 다운경로]
  • maven 파일 선택하기

소스코드

  • java File

    /*
     * Copyright 2015 Fabian Hueske / Vasia Kalavri
     *
     * Licensed under the Apache License, Version 2.0 (the "License");
     * you may not use this file except in compliance with the License.
     * You may obtain a copy of the License at
     *
     *  http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS,
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     * See the License for the specific language governing permissions and
     * limitations under the License.
     */
    package io.github.streamingwithflink.chapter1;
    
    import io.github.streamingwithflink.util.SensorReading;
    import io.github.streamingwithflink.util.SensorSource;
    import io.github.streamingwithflink.util.SensorTimeAssigner;
    import org.apache.flink.streaming.api.TimeCharacteristic;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    import org.apache.flink.util.Collector;
    
    public class AverageSensorReadings {
    
        /**
         * main() defines and executes the DataStream program.
         *
         * @param args program arguments
         * @throws Exception
         */
        public static void main(String[] args) throws Exception {
    
            // set up the streaming execution environment
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            // use event time for the application
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            // configure watermark interval
            env.getConfig().setAutoWatermarkInterval(1000L);
    
            // ingest sensor stream
            DataStream<SensorReading> sensorData = env
                // SensorSource generates random temperature readings
                .addSource(new SensorSource())
                // assign timestamps and watermarks which are required for event time
                .assignTimestampsAndWatermarks(new SensorTimeAssigner());
    
            DataStream<SensorReading> avgTemp = sensorData
                // convert Fahrenheit to Celsius using and inlined map function
                .map( r -> new SensorReading(r.id, r.timestamp, (r.temperature - 32) * (5.0 / 9.0)))
                // organize stream by sensor
                .keyBy(r -> r.id)
                // group readings in 1 second windows
                .timeWindow(Time.seconds(1))
                // compute average temperature using a user-defined function
                .apply(new TemperatureAverager());
    
            // print result stream to standard out
            avgTemp.print();
    
            // execute application
            env.execute("Compute average sensor temperature");
        }
    
        /**
         *  User-defined WindowFunction to compute the average temperature of SensorReadings
         */
        public static class TemperatureAverager implements WindowFunction<SensorReading, SensorReading, String, TimeWindow> {
    
            /**
             * apply() is invoked once for each window.
             *
             * @param sensorId the key (sensorId) of the window
             * @param window meta data for the window
             * @param input an iterable over the collected sensor readings that were assigned to the window
             * @param out a collector to emit results from the function
             */
            @Override
            public void apply(String sensorId, TimeWindow window, Iterable<SensorReading> input, Collector<SensorReading> out) {
    
                // compute the average temperature
                int cnt = 0;
                double sum = 0.0;
                for (SensorReading r : input) {
                    cnt++;
                    sum += r.temperature;
                }
                double avgTemp = sum / cnt;
    
                // emit a SensorReading with the average temperature
                out.collect(new SensorReading(sensorId, window.getEnd(), avgTemp));
            }
        }
    }
    
  • pom.xml

    <!--
    Licensed to the Apache Software Foundation (ASF) under one
    or more contributor license agreements.  See the NOTICE file
    distributed with this work for additional information
    regarding copyright ownership.  The ASF licenses this file
    to you under the Apache License, Version 2.0 (the
    "License"); you may not use this file except in compliance
    with the License.  You may obtain a copy of the License at
    
      http://www.apache.org/licenses/LICENSE-2.0
    
    Unless required by applicable law or agreed to in writing,
    software distributed under the License is distributed on an
    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    KIND, either express or implied.  See the License for the
    specific language governing permissions and limitations
    under the License.
    -->
    <project xmlns="http://maven.apache.org/POM/4.0.0" 
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
    
     <groupId>io.github.streamingwithflink</groupId>
     <artifactId>examples-java</artifactId>
     <version>1.0</version>
     <packaging>jar</packaging>
    
     <name>Java Examples for Stream Processing with Apache Flink</name>
     <url>http://streaming-with-flink.github.io/examples</url>
    
     <properties>
      <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
      <flink.version>1.7.1</flink.version>
      <java.version>1.8</java.version>
      <scala.binary.version>2.12</scala.binary.version>
      <maven.compiler.source>${java.version}</maven.compiler.source>
      <maven.compiler.target>${java.version}</maven.compiler.target>
     </properties>
    
     <dependencies>
      <!-- Apache Flink dependencies -->
      <!-- These dependencies are provided, because they should not be packaged into the JAR file. -->
      <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-java</artifactId>
       <version>${flink.version}</version>
       <scope>provided</scope>
      </dependency>
      <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
       <version>${flink.version}</version>
       <scope>provided</scope>
      </dependency>
    
      <!-- runtime-web dependency is need to start web UI from IDE -->
                    <dependency>
                            <groupId>org.apache.flink</groupId>
                            <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
                            <version>${flink.version}</version>
                            <scope>provided</scope>
                    </dependency>
    
      <!-- queryable-state dependencies are needed for respective examples -->
                    <dependency>
                            <groupId>org.apache.flink</groupId>
                            <artifactId>flink-queryable-state-runtime_${scala.binary.version}</artifactId>
                            <version>${flink.version}</version>
                    </dependency>
                    <dependency>
                            <groupId>org.apache.flink</groupId>
                            <artifactId>flink-queryable-state-client-java_${scala.binary.version}</artifactId>
                            <version>${flink.version}</version>
                    </dependency>
    
      <!--
    
                    Derby is used for a sink connector example.
                    Example only works in local mode, i.e, it is not possible to submit it to a running cluster.
                    The dependency is set to provided to reduce the size of the JAR file.
                    -->
                    <dependency>
                            <groupId>org.apache.derby</groupId>
                            <artifactId>derby</artifactId>
                            <version>10.13.1.1</version>
                            <scope>provided</scope>
                    </dependency>
    
                    <!-- Logging -->
                    <dependency>
                            <groupId>org.slf4j</groupId>
                            <artifactId>slf4j-log4j12</artifactId>
                            <version>1.7.25</version>
                            <scope>runtime</scope>
                    </dependency>
                    <dependency>
                            <groupId>log4j</groupId>
                            <artifactId>log4j</artifactId>
                            <version>1.2.17</version>
                            <scope>runtime</scope>
                    </dependency>
     </dependencies>
    
     <build>
      <plugins>
       <!-- Java Compiler -->
       <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.1</version>
        <configuration>
         <source>${java.version}</source>
         <target>${java.version}</target>
        </configuration>
       </plugin>
    
       <!-- We use the maven-shade plugin to create a fat jar that contains all necessary dependencies. -->
       <!-- Change the value of <mainClass>...</mainClass> if your program entry point changes. -->
       <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-shade-plugin</artifactId>
        <version>3.0.0</version>
        <executions>
         <!-- Run shade goal on package phase -->
         <execution>
          <phase>package</phase>
          <goals>
           <goal>shade</goal>
          </goals>
          <configuration>
           <artifactSet>
            <excludes>
             <exclude>org.apache.flink:force-shading</exclude>
             <exlude>org.apache.flink:flink-shaded-netty</exlude>
                                                                            <exlude>org.apache.flink:flink-shaded-guava</exlude>
             <exclude>com.google.code.findbugs:jsr305</exclude>
             <exclude>org.slf4j:*</exclude>
             <exclude>log4j:*</exclude>
            </excludes>
           </artifactSet>
           <filters>
            <filter>
             <!-- Do not copy the signatures in the META-INF folder.
             Otherwise, this might cause SecurityExceptions when using the JAR. -->
             <artifact>*:*</artifact>
             <excludes>
              <exclude>META-INF/*.SF</exclude>
              <exclude>META-INF/*.DSA</exclude>
              <exclude>META-INF/*.RSA</exclude>
             </excludes>
            </filter>
           </filters>
           <transformers>
            <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
             <mainClass>io.github.streamingwithflink.StreamingJob</mainClass>
            </transformer>
           </transformers>
          </configuration>
         </execution>
        </executions>
       </plugin>
      </plugins>
    
      <pluginManagement>
       <plugins>
    
        <!-- This improves the out-of-the-box experience in Eclipse by resolving some warnings. -->
        <plugin>
         <groupId>org.eclipse.m2e</groupId>
         <artifactId>lifecycle-mapping</artifactId>
         <version>1.0.0</version>
         <configuration>
          <lifecycleMappingMetadata>
           <pluginExecutions>
            <pluginExecution>
             <pluginExecutionFilter>
              <groupId>org.apache.maven.plugins</groupId>
              <artifactId>maven-shade-plugin</artifactId>
              <versionRange>[3.0.0,)</versionRange>
              <goals>
               <goal>shade</goal>
              </goals>
             </pluginExecutionFilter>
             <action>
              <ignore/>
             </action>
            </pluginExecution>
            <pluginExecution>
             <pluginExecutionFilter>
              <groupId>org.apache.maven.plugins</groupId>
              <artifactId>maven-compiler-plugin</artifactId>
              <versionRange>[3.1,)</versionRange>
              <goals>
               <goal>testCompile</goal>
               <goal>compile</goal>
              </goals>
             </pluginExecutionFilter>
             <action>
              <ignore/>
             </action>
            </pluginExecution>
           </pluginExecutions>
          </lifecycleMappingMetadata>
         </configuration>
        </plugin>
       </plugins>
      </pluginManagement>
     </build>
    
     <!-- This profile helps to make things run out of the box in IntelliJ -->
     <!-- Its adds Flink's core classes to the runtime class path. -->
     <!-- Otherwise they are missing in IntelliJ, because the dependency is 'provided' -->
     <profiles>
      <profile>
       <id>add-dependencies-for-IDEA</id>
    
       <activation>
        <property>
         <name>idea.version</name>
        </property>
       </activation>
    
       <dependencies>
        <dependency>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-java</artifactId>
         <version>${flink.version}</version>
         <scope>compile</scope>
        </dependency>
        <dependency>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
         <version>${flink.version}</version>
         <scope>compile</scope>
        </dependency>
        <dependency>
                                            <groupId>org.apache.flink</groupId>
                                            <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
                                            <version>${flink.version}</version>
                                            <scope>compile</scope>
                                    </dependency>
                                    <dependency>
                                            <groupId>org.apache.derby</groupId>
                                            <artifactId>derby</artifactId>
                                            <version>10.13.1.1</version>
                                            <scope>compile</scope>
                                    </dependency>
       </dependencies>
      </profile>
     </profiles>
    
    </project>
    
    

JAR 파일 만들기

프로젝트 우클릭 → 모듈설정 → 아티팩트 설정 → 추가 → JAR file → 종속성 포함 모듈 설정

위의 명령대로 실행하면 아래의 화면이 나온다.

image
image

메인 클래스 오른쪽에 폴더를 누르면 우리가 실행 할 Main 파일이 자동으로 등록 되어 있디.

설정이 완료되면 ‘확인’을 눌러주고 팝업창을 닫아주자.

이제 JAR 파일을 말아주어야 한다. 설정한 아티팩트대로 JAR가 빌드 된다. 아까 우리가 아티팩트를 설정 했기 때문에 빌드 메뉴의 아티팩트 빌드가 활성화 되어있을 것이다.

빌드 → 아티팩트 빌드

빌드가 완료된다면 우리가 지정한 경로에 JAR 파일이 생성된다 (특별한 설정을 하지 않았기 때문에 out/artifacts/examples_jar 폴더 아래에 생성 된다.

Flink Job 등록하기

이제 만들어 준 Flink Job을 Docker로 설정 한 Flink 서버에 넣어 주어야 한다.

왼쪽 메뉴에 가장 아래에 있는 “Submit new Job”을 눌러주자

image
image

Add New 를 누르면, 우리가 만든 JAR 파일을 하나의 Job으로 등록 할 수 있다.

Add New를 누르고, JAR 파일을 선택해서 Upload하면 아래의 화면처럼 JOB이 등록된다.

image
image

이제 실행 할 수 있도록 해당하는 job을 눌러주고 실행을 위해 submit을 눌러주세요

image
image

Submit이 완료되면 아래의 화면처럼 태스크가 실행 되는 것을 확인 할 수 있다.

image
image

OverView 메뉴에서도 우리가 실행 한 작업들이 잘 보인다

image
image

아직 완료하지 못한 것

docker가 아닌 로컬에서 Flink를 실행하고 job을 올리면 잘 보이는 Flink log가 Docker 상에서는 보이지 않는다. 아직 이유를 모른다.

image
image

"Internal server error.","<Exception on server side:org.apache.flink.util.FlinkException: The file STDOUT is not available on the TaskExecutor…… "

요런 에러가 뜨는데 해결 해야 할 Task로 남겨두고 공부 해 봐야겠다..


Tags

#Flink#Stream#Docker#Apache Flink#Flink Job
Previous Article
Kafka로 메시지와 이벤트 처리하기 - (1) Kafka 세팅하기
울이

울이

개발자 울이

경험하고 사용하고 개선하는 것, 그리고 소통하면서 성장하는 것을 좋아합니다.

Expertise

Backend
Frontend
Data Engineering

Social Media

githublinkedinwebsiteinstagram

Related Posts

AWS Kinesis python으로 다뤄보기
AWS Kinesis python으로 다뤄보기
December 30, 2021
1 min
Docker로 작성한 서버, Debug모드 적용하기
Backend Engineering
Docker로 작성한 서버, Debug모드 적용하기
January 12, 2022
1 min
© 2022, All Rights Reserved.

Quick Links

Home

Social Media