flink1.12.x 打包pom.xml，yarn方式运行（socket案例）

作者：马育民 • • 阅读：10164

修改pom.xml

不包含flink自带包

flink自带的jar包，在依赖上要添加 <scope>provided</scope>，打包时，不要包含这些包

<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_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>1.7.30</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.30</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>log4j</groupId>
    <artifactId>log4j</artifactId>
    <version>1.2.17</version>
    <scope>provided</scope>
</dependency>

注意

flink-runtime-web_2.12 要设置 <scope>provided</scope>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-runtime-web_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>

注意

org.apache.bahir 不要用，因为是 scala 2.11 版本，与 flink 中的scala 2.12 版不匹配

<dependency>
    <groupId>org.apache.bahir</groupId>
    <artifactId>flink-connector-redis_2.11</artifactId>
    <version>1.0</version>
</dependency>

添加打包插件

<build>
    <plugins>
        <plugin>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.7.0</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>3.1.1</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                    <configuration>
                        <transformers>
                            <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                <!--   修改主类-->
                                <mainClass></mainClass>
                            </transformer>
                            <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                <resource>reference.conf</resource>
                            </transformer>
                        </transformers>
                        <relocations>
                            <relocation>
                                <pattern>org.codehaus.plexus.util</pattern>
                                <shadedPattern>org.shaded.plexus.util</shadedPattern>
                                <excludes>
                                    <exclude>org.codehaus.plexus.util.xml.Xpp3Dom</exclude>
                                    <exclude>org.codehaus.plexus.util.xml.pull.*</exclude>
                                </excludes>
                            </relocation>
                        </relocations>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

完整依赖

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <flink.version>1.12.0</flink.version>
</properties>
<dependencies>
    <!--java-->
    <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_2.12</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.12</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.30</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.30</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
        <scope>provided</scope>
    </dependency>
    <!-- kafka Dependency -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-files</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-runtime-web_2.12</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-jdbc_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.47</version>
    </dependency>
    <!--<dependency>-->
    <!--    <groupId>org.apache.bahir</groupId>-->
    <!--    <artifactId>flink-connector-redis_2.11</artifactId>-->
    <!--    <version>1.0</version>-->
    <!--</dependency>-->

    <dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>2.9.0</version>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.7.0</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>3.1.1</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                    <configuration>
                        <transformers>
                            <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                <!--   修改主类-->
                                <mainClass></mainClass>
                            </transformer>
                            <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                <resource>reference.conf</resource>
                            </transformer>
                        </transformers>
                        <relocations>
                            <relocation>
                                <pattern>org.codehaus.plexus.util</pattern>
                                <shadedPattern>org.shaded.plexus.util</shadedPattern>
                                <excludes>
                                    <exclude>org.codehaus.plexus.util.xml.Xpp3Dom</exclude>
                                    <exclude>org.codehaus.plexus.util.xml.pull.*</exclude>
                                </excludes>
                            </relocation>
                        </relocations>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

java 类

package map;


import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class TestSocket {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setRuntimeMode(RuntimeExecutionMode.BATCH);//批模式
//        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);//流模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
        DataStream<String> dataStream = env.socketTextStream("hadoop1", 9999);
        SingleOutputStreamOperator<Tuple2<String, Integer>> soso=dataStream.flatMap(new Splitter());
        KeyedStream<Tuple2<String, Integer>,String> ks=soso.keyBy(new keyByImpl());//流处理中的分组
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
        SingleOutputStreamOperator<Tuple2<String, Integer>> soso2=ks.sum(1);//按照第二个元素聚合
        soso2.print();
        //启动并等待程序结束，可传入任务名
        env.execute("Window WordCount");
    }
    /**
     * 拆分、并记为1，每个元素是元祖，格式： (单词,1)
     * FlatMapFunction<String, Tuple2<String, Integer>> 泛型解释：
     *      String：传入的是字符串
     *      Tuple2<String, Integer>，集合的泛型。对字符串进行处理，封装成Tuple类型，格式：(单词,1)，然后放到该集合中
     */
    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        /**
         *
         * @param sentence 表示每一行数据
         * @param out  传出的集合
         */
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }
    /**
     根据第一个元素（即：单词）进行分组
     KeySelector<Tuple2<String, Integer>,String> 泛型解释：
     Tuple2<String, Integer>：传入的是 Tuple类型，也就是上面封装的Tuple，格式： (单词,1)
     String：传出类型，也就是Tuple中的单词，根据该单词进行分组
     */
    public static class keyByImpl implements KeySelector<Tuple2<String, Integer>,String>{
        /**
         *
         * @param t：传入的元祖，格式：(单词,1)
         * @return
         * @throws Exception
         */
        @Override
        public String getKey(Tuple2<String, Integer> t) throws Exception {
            return t.f0;
        }
    }
}

log4j.properties

在 resources 目录下创建 log4j.properties 文件，内容如下：

# DEBUG 级别，即：所有信息都会输出
log4j.rootLogger=INFO, stdout
# 配置 stdout，输出到控制台
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

注意：不加此文件会显示警告

测试

打包

双击 maven 的 package，打包，结果有2个 jar 文件，一个文件比较大，这个文件就是打包好的文件

启动 socket

登录 hadoop1，执行下面命令：

nc -lk 9999

上传

略

per-job cluster模式启动

/program/flink-1.12.5/bin/flink run -m yarn-cluster -ytm 1000 -yjm 1000 -ys 1  -c  map.TestSocket    /program/flink-1.0-SNAPSHOT.jar

看到下图表示启动成功：

输入数据

在 socket 窗口输入下面内容：

haha hehe
xixi haha xixi

查看结果

原文出处：/show_1IX4DKrWNgzq.html

吉ICP备15003043号