说明

读取 HDFS 文件，并将结果写入到 HDFS 文件中
将编写好的程序打成 jar 包，上传到 hadoop1，通过 flink run 命令执行

java代码

package top.malaoshi;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;

public class TestDataStream2_lambda2_hdfs {

    public static void main(String[] args) throws Exception {
        ParameterTool pt=ParameterTool.fromArgs(args);



        String output="hdfs://hadoop1:8020/flink_result/result_";
        if(pt.has("output")){
            output=pt.get("output");
        }
        System.out.println("结果保存路径是："+output);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setRuntimeMode(RuntimeExecutionMode.BATCH);//批模式
//        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);//流模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断

        DataStream<String> dataStream=null;
        if(pt.has("input")){
            String input=pt.get("input");
            System.out.println("要读取的文件路径是："+input);
            dataStream = env.readTextFile(input);
        }else{
            System.out.println("没有传入要读取的文件路径，处理默认字符串！");
            dataStream = env.fromElements(
                    "hadoop hdfs",
                    "hadoop yarn",
                    "hadoop flink",
                    "flink yarn");
        }


        DataStream ds=dataStream.flatMap(
                        (String value, Collector<String> out)
                                -> Arrays.stream(value.split(" ")).forEach( out :: collect)).returns(Types.STRING)//根据空格拆分
                .map( (String value ) -> Tuple2.of( value ,1)).returns(Types.TUPLE(Types.STRING,Types.INT))//将每个元素的格式转为： (单词,1)
                .keyBy(value -> value.f0)//流处理中的分组，根据第一个元素（即：单词）进行分组
                .sum(1);//聚合
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))

//        dataStream.print();
        //如果执行报hdfs权限错误，执行命令：hadoop fs -chmod -R 777 /
        System.setProperty("HADOOP_USER_NAME","root");
        SimpleDateFormat sdf=new SimpleDateFormat("yyyyMMdd_HHmmss");
        String datetime=sdf.format(new Date());
        /*
        写入hdfs
         */
        ds.writeAsText(output+datetime)
                .setParallelism(1);//并行度，生成1个文件
        //启动并等待程序结束，可传入任务名
        env.execute("Window WordCount");
    }

}

idea 打包

执行 maven 的 package 命令，包名改为：test_flink.jar

上传

将 test_flink.jar 包上传到 hadoop1 的 /program 目录下

执行

默认执行：

/program/flink-1.12.5/bin/flink run -c top.malaoshi.TestDataStream2_lambda2_hdfs /program/test_flink.jar

读取指定的 HDFS 文件，并将结果写入到指定 HDFS 路径中：

/program/flink-1.12.5/bin/flink run \
-c top.malaoshi.TestDataStream2_lambda2_hdfs \
/program/test_flink.jar \
--input hdfs://hadoop1:8020//test/data.txt  \
--output hdfs://hadoop1:8020/test_result/2

查看

说明

java代码

idea 打包

上传

执行

查看

查看 8081 服务

查看 8082 服务

查看 hdfs 结果文件