DolphinScheduler 传递参数到flink程序中_flink 提交命令传日期参数

1. 通过DolphinScheduler 将参数传递到flink 程序中。

2.定义工作流程

3. flink 代码程序：测试代码

package com.wudl.flink.instance;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.types.Row;
import org.apache.hadoop.mapreduce.Job;

import java.util.stream.Stream;

/**
 * @ClassName : WriteToMysql
 * @Description : flink-jdbc
 * @Author :wudl
 * @Date: 2021-08-01 20:36
 */

public class WriteToMysql {

    public static void main(String[] args) throws Exception {
        System.out.println("----------args" + args);
        System.out.println("----------args" + args.toString());
        for (int i = 0; i < args.length; i++) {
            System.out.println("-----i---" + args[i]);
        }

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        System.out.println("------------------------");
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        ParameterTool parameter = ParameterTool.fromSystemProperties();
        System.out.println("88888" + parameter.getNumberOfParameters());
        System.out.println("888884545555" + parameter.toString());
        System.out.println("fromSystemProperties--------" + parameter.get("programType"));
        System.out.println("fromSystemProperties2222-----------" + parameter.get("java"));
        System.out.println("fromSystemProperties2222-333----------" + parameter.get("mainClass"));
        System.out.println("88888" + parameter.getNumberOfParameters());
        System.out.println("88888" + parameter.getNumberOfParameters());
        System.out.println("88888" + parameter.getNumberOfParameters());

        System.out.println("打印参数----" + parameterTool.toString());
        System.out.println("********111**********" + parameterTool.get("java"));
        System.out.println("********111programType**********" + parameterTool.get("programType"));
        System.out.println("********111resourceList**********" + parameterTool.get("resourceList"));
        System.out.println("*******2222***********" + parameterTool.getProperties().getProperty("java"));
        System.out.println("*******3333***********" + parameterTool.getProperties().get("java"));
        System.out.println("*******3333***********" + parameterTool.getProperties().toString());
        System.out.println("*******3333***********" + parameterTool.getProperties());
        System.out.println("*******3333***********" + parameterTool.getProperties());

        System.out.println("");
//        env.getConfig().setGlobalJobParameters(parameterTool);

        DataSet<Row> inputMysql = env.createInput(JDBCInputFormat.buildJDBCInputFormat()
                //配置数据库连接信息
                .setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl("jdbc:mysql://192.168.1.162:3306/test")
                .setUsername("root")
                .setPassword("123456")
                .setQuery("select * from testtable")
                //设置查询的列的类型，根据实际情况定
                .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO))
                .finish());


        inputMysql.print();
        System.out.println("************");
//
//        inputMysql.map(new MapFunction<Row, String>() {
//            @Override
//            public String map(Row value) throws Exception {
//                System.out.println(value);
//                return value.toString();
//            }
//        }).print();

        //mysql连接
        String driverClass = "com.mysql.jdbc.Driver";
        String dbUrl = "jdbc:mysql://192.168.1.162:3306/test?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true ";
        String userNmae = "root";
        String passWord = "123456";
        String sql = "insert into testtable02 (id,mamer) values (?,?)";
        //将数据写入mysql数据库
        inputMysql.output(JDBCOutputFormat.buildJDBCOutputFormat()
                .setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl("jdbc:mysql://192.168.1.162:3306/test")
                .setUsername("root")
                .setPassword("123456")
                .setQuery(sql)
                .finish());
        //这个必须要执行，否则看不到结果
        Job job = Job.getInstance();
        env.execute("insert into mysql");
    }

//    class FilterGenreWithGlobalEnv extends RichFilterFunction<Tuple3<Long, String, String>> {
//        @Override
//        public boolean filter(Tuple3<Long, String, String> movie) throws Exception {
//            String[] genres = movie.f2.split(",");
//            // Get global parameters
//            ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
//            // Read parameter
//            String genre = parameterTool.get("genre");
//            return Stream.of(genres).anyMatch(g -> g.equals(genre));
//        }
//    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

4. 日志：

[INFO] 2021-08-03 00:31:22.695  - [taskAppId=TASK-16-66-76]:[115] - create dir success /tmp/dolphinscheduler/exec/process/5/16/66/76
[INFO] 2021-08-03 00:31:23.484  - [taskAppId=TASK-16-66-76]:[70] - flink task params {"mainArgs":" -path kkkkkkkkkkkkkkkkkk,hhhhhhhhhhhhhhhhhh","programType":"JAVA","mainClass":"com.wudl.flink.instance.WriteToMysql","deployMode":"local","appName":"","parallelism":1,"taskManagerMemory":"2G","mainJar":{"id":9},"slot":1,"flinkVersion":"<1.10","taskManager":"2","localParams":[],"others":"","resourceList":[{"res":"wudl-2021.jar","name":"wudl-2021.jar","id":9}],"jobManagerMemory":"1G"}
[INFO] 2021-08-03 00:31:23.489  - [taskAppId=TASK-16-66-76]:[90] - param Map : {}
[INFO] 2021-08-03 00:31:23.489  - [taskAppId=TASK-16-66-76]:[93] - param args :  -path kkkkkkkkkkkkkkkkkk,hhhhhhhhhhhhhhhhhh
[INFO] 2021-08-03 00:31:23.489  - [taskAppId=TASK-16-66-76]:[110] - flink task args : [flink, run]
[INFO] 2021-08-03 00:31:23.489  - [taskAppId=TASK-16-66-76]:[117] - flink task command : flink run -p 1 -sae -c com.wudl.flink.instance.WriteToMysql wudl-2021.jar  -path kkkkkkkkkkkkkkkkkk,hhhhhhhhhhhhhhhhhh
[INFO] 2021-08-03 00:31:23.490  - [taskAppId=TASK-16-66-76]:[87] - tenantCode user:hdfs, task dir:16_66_76
[INFO] 2021-08-03 00:31:23.490  - [taskAppId=TASK-16-66-76]:[92] - create command file:/tmp/dolphinscheduler/exec/process/5/16/66/76/16_66_76.command
[INFO] 2021-08-03 00:31:23.490  - [taskAppId=TASK-16-66-76]:[111] - command : #!/bin/sh
BASEDIR=$(cd `dirname $0`; pwd)
cd $BASEDIR
source /ds1.3.6/conf/env/dolphinscheduler_env.sh
flink run -p 1 -sae -c com.wudl.flink.instance.WriteToMysql wudl-2021.jar  -path kkkkkkkkkkkkkkkkkk,hhhhhhhhhhhhhhhhhh
[INFO] 2021-08-03 00:31:23.491  - [taskAppId=TASK-16-66-76]:[327] - task run command:
sudo -u hdfs sh /tmp/dolphinscheduler/exec/process/5/16/66/76/16_66_76.command
[INFO] 2021-08-03 00:31:23.491  - [taskAppId=TASK-16-66-76]:[208] - process start, process id is: 28836
[INFO] 2021-08-03 00:31:24.496  - [taskAppId=TASK-16-66-76]:[129] -  -> Setting HBASE_CONF_DIR=/etc/hbase/conf because no HBASE_CONF_DIR was set.
[INFO] 2021-08-03 00:31:25.497  - [taskAppId=TASK-16-66-76]:[129] -  -> ----------args[Ljava.lang.String;@6f44a157
	----------args[Ljava.lang.String;@6f44a157
	-----i----path
	-----i---kkkkkkkkkkkkkkkkkk,hhhhhhhhhhhhhhhhhh
	------------------------
	8888858
	888884545555org.apache.flink.api.java.utils.ParameterTool@c01d802a
	fromSystemProperties--------null
	fromSystemProperties2222-----------null
	fromSystemProperties2222-333----------null
	8888858
	8888858
	8888858
	打印参数----org.apache.flink.api.java.utils.ParameterTool@7cf5ec2d
	********111**********null
	********111programType**********null
	********111resourceList**********null
	*******2222***********null
	*******3333***********null
	*******3333***********{path=kkkkkkkkkkkkkkkkkk,hhhhhhhhhhhhhhhhhh}
	*******3333***********{path=kkkkkkkkkkkkkkkkkk,hhhhhhhhhhhhhhhhhh}
	*******3333***********{path=kkkkkkkkkkkkkkkkkk,hhhhhhhhhhhhhhhhhh}
	
[INFO] 2021-08-03 00:31:29.509  - [taskAppId=TASK-16-66-76]:[129] -  -> Job has been submitted with JobID d81184dbdcaf2feffed43e42b90361fd
[INFO] 2021-08-03 00:31:31.511  - [taskAppId=TASK-16-66-76]:[129] -  -> Program execution finished
	Job with JobID d81184dbdcaf2feffed43e42b90361fd has finished.
	Job Runtime: 1335 ms
	Accumulator Results: 
	- 8aa67122e79dada47a718555a8fe18fd (java.util.ArrayList) [6 elements]
	
	
	1,hdfs
	2,hive
	3,111
	4,222222222222222
	5,33333333333333
	6,44444444444444444
	************
[INFO] 2021-08-03 00:31:37.706  - [taskAppId=TASK-16-66-76]:[217] - process has exited, execute path:/tmp/dolphinscheduler/exec/process/5/16/66/76, processId:28836 ,exitStatusCode:0
[INFO] 2021-08-03 00:31:38.514  - [taskAppId=TASK-16-66-76]:[129] -  -> Job has been submitted with JobID b27f2f535c664cf95eeb4b9608ae3fa8
	Program execution finished
	Job with JobID b27f2f535c664cf95eeb4b9608ae3fa8 has finished.
	Job Runtime: 1960 ms
	

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

5. 核心点就在将参数伪装成 -path 输出路径, 因为path 不影响命令的执行。

可以看出flink 执行的命令：
flink run -p 1 -sae -c com.wudl.flink.instance.WriteToMysql wudl-2021.jar -path kkkkkkkkkkkkkkkkkk,hhhhhhhhhhhhhhhhhh

6. 可以采用接口调用 save 接口调用伪装参数