DolphinScheduler 传递参数到flink程序中

作者: wudl | 来源:发表于2021-08-03 00:40 被阅读0次

DolphinScheduler 传递参数到flink程序中
uni-app路由传参(传递对象 )
DolphinScheduler 调用flink以及api的使
Flink最佳实践
列表进入详情页面传参的两种方式总结 query params
scrapy传参，动态创建items
flink function类传递参数(Passing Para
微信小程序原生开发技巧及跳坑心得
Flink的并发执行(Parallel Execution)
Java 与值传递

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

2.定义工作流程

flink-传递参数.png

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));
//        }
//    }

}

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

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

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

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

DolphinScheduler 传递参数到flink程序中
1. 通过DolphinScheduler 将参数传递到flink 程序中。 2.定义工作流程 3. flink ...
uni-app路由传参(传递对象 )
uni-app路由传参(传递对象 )：遇到需要在navigaor组件中传递对象参数到下一个页面的需求，传递对象时...
DolphinScheduler 调用flink以及api的使
1.ds 调用flink 的使用官方文档连接：https://dolphinscheduler.apache....
Flink最佳实践
解析命令行参数和在Flink应用程序中传递参数几乎所有的Flink应用程序，包括批处理和流处理，都依赖于外部配置...
列表进入详情页面传参的两种方式总结 query params
在开发中，我们往往遇到需要跳转页面传参来传递数据，比如列表传递参数到详情页面；详情页面的url为：我们可以看到传...
scrapy传参，动态创建items
scrapy从启动文件传递参数到spiders start.py传参 spiders.py 接收参数创建动态it...
flink function类传递参数(Passing Para
在flink的function类中传递参数，对于flink datastream和dataset是不同的，对于da...
微信小程序原生开发技巧及跳坑心得
常用功能 1、事件带参在微信小程序中，给事件的回调函数内部传递参数不是通过函数调用传递实参的方式传递的，而是通过...
Flink的并发执行(Parallel Execution)
Flink的并发执行本章描述如何在Flink中配置程序的并发执行，一个Flink程序可以由不同的task(如:t...
Java 与值传递
问：什么是值传递？什么是引用传递？为什么说 Java 中只有值传递？一、值传递与引用传递实参与形参：实际参...