hamawhitegg / flink-sql-lineage Goto Github PK

CREATE TABLE `sink_table`(`agent_id` STRING, `room_id` STRING, `application_id` STRING, `type` STRING, `begin_time` BIGINT, `end_time` BIGINT) WITH(...)

CREATE TABLE `source_table`(`agent_id` STRING, `room_id` STRING, `create_time` BIGINT, `type` STRING, `application_id` STRING, `connect_time` BIGINT, `row_time` AS PROCTIME()) WITH(...)

insert into sink_table (agent_id,room_id,application_id,type,begin_time,end_time) 
select agent_id,room_id,application_id,type,begin_time,end_time 
	from source_table 
	match_recognize (partition by agent_id,room_id,application_id order by row_time 
	measures AF.type as type,last(BF.create_time) as begin_time,last(AF.create_time) as end_time 
	one row per match 
	after match SKIP PAST LAST ROW 
	pattern (BF+ AF) WITHIN INTERVAL '1' HOUR 
	define BF as BF.type = 'assign',AF as AF.type = 'pick_up' ) as T

@Slf4j
public class SimpleTest extends AbstractBasicTest {

    @Before
    public void createTable() {
        createMysqlUser();
        createOdsUser();
    }

    @Test
    public void concat() {
        String sql = "insert into mysql_user " +
                "select id, birthday, concat(first_name, last_name) as full_name " +
                "from ods_user";
        List<Result> actualList = context.parseFieldLineage(sql);
        log.info("Linage Result: ");
        actualList.forEach(e -> log.info(e.toString()));
    }

    protected void createMysqlUser() {
        context.execute("DROP TABLE IF EXISTS mysql_user ");
        context.execute("CREATE TABLE IF NOT EXISTS mysql_user (" +
                "       id                        BIGINT           ," +
                "       birthday                  TIMESTAMP(3)     ," +
                "       full_name                 STRING            " +
                ") WITH ( " +
                "       'connector' = 'jdbc'                 ," +
                "       'url'       = 'jdbc:mysql://127.0.0.1:3306/demo?useSSL=false&characterEncoding=UTF-8'," +
                "       'username'  = 'root'                 ," +
                "       'password'  = 'xxx'          ," +
                "       'table-name'= 'mysql_user' " +
                ")"
        );
    }

    protected void createOdsUser() {
        context.execute("DROP TABLE IF EXISTS ods_user ");
        context.execute("CREATE TABLE IF NOT EXISTS ods_user (" +
                "       id                        BIGINT           ," +
                "       birthday                  TIMESTAMP(3)     ," +
                "       first_name                STRING           ," +
                "       last_name                 STRING           ," +
                "       company_name              STRING           " +
                ") WITH ( " +
                "       'connector' = 'jdbc'                 ," +
                "       'url'       = 'jdbc:mysql://127.0.0.1:3306/demo?useSSL=false&characterEncoding=UTF-8'," +
                "       'username'  = 'root'                 ," +
                "       'password'  = 'xxx'          ," +
                "       'table-name'= 'ods_user' " +
                ")"
        );
    }
}

    @Test
    public void testInsertSelectUUID() {
        String sql = "INSERT INTO dwd_hudi_users (id, name, birthday) " +
                "SELECT " +
                "   ROW_NUMBER() OVER (ORDER BY ts DESC) as id," +
                "   UUID() as name," +
                "   birthday " +
                "FROM" +
                "   ods_mysql_users";
    
        String[][] expectedArray = {
                {"ods_mysql_users", "ts", "dwd_hudi_users", "id",
                        "ROW_NUMBER() OVER (ORDER BY ts DESC NULLS LAST)"},
                {"ods_mysql_users", "name", "dwd_hudi_users", "name"},
                {"ods_mysql_users", "ts", "dwd_hudi_users", "birthday"}
        };
        
        analyzeLineage(sql, expectedArray);
    }

-- source1
DROP TABLE IF EXISTS `dwd_status_log`;
CREATE TABLE `dwd_status_log`
(
    mechine STRING,
    timenow STRING,
    `value` STRING,
    pre_value STRING,
    PRIMARY KEY (timenow) NOT ENFORCED
)  
WITH (
    'connector' = 'kafka',
    'topic' = 'dwd_status_log',
    'scan.startup.mode' = 'earliest-offset',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'debezium-json'
);

-- source2
DROP TABLE IF EXISTS dim_equipment_info;
CREATE TABLE IF NOT EXISTS dim_equipment_info (
    id STRING 
    ,station_type STRING 
    ,equipment_no STRING 
    ,emp_name STRING 
    ,emp_phone STRING 
    ,vendor STRING 
    ,site STRING 
    ,`floor` STRING 
    ,line STRING 
    ,model STRING 
    ,stage STRING 
    ,id_value STRING 
    ,emp_info STRING
    ,PRIMARY KEY ( id ) NOT ENFORCED
) 
WITH (
   'connector' = 'jdbc',
   'url' = 'jdbc:mysql://localhost:4000/data_warehouse_p_10046?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&nullCatalogMeansCurrent=true',
   'table-name' = 'dim_equipment_info',
   'username' = 'root',
   'password' = 'xxx'
);


-- sink
DROP TABLE IF EXISTS dws_alarm_status_count;
CREATE TABLE IF NOT EXISTS dws_alarm_status_count (
    mechine_id STRING 
    ,mechine_name STRING 
    ,trigger_date STRING 
    ,alarm_status_count BIGINT 
    ,PRIMARY KEY ( mechine_id,trigger_date ) NOT ENFORCED
) 
WITH (
    'connector' = 'kafka',
    'topic' = 'dws_alarm_status_count',
    'scan.startup.mode' = 'earliest-offset',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'debezium-json'
);

-- trans
INSERT INTO 
    dws_alarm_status_count 
SELECT
    dsl.mechine AS mechine_id,
    dei.equipment_no AS mechine_name,
    DATE_FORMAT(dsl.timenow, 'yyyy-MM-dd') AS `trigger_date`,
    COUNT(*) AS alarm_status_count
FROM 
    dwd_status_log dsl 
LEFT JOIN 
    dim_equipment_info dei 
ON 
    dsl.mechine = dei.id 
WHERE 
    dsl.`value` = '4' 
GROUP BY 
DATE_FORMAT(dsl.timenow, 'yyyy-MM-dd'), dsl.mechine,dei.equipment_no;

public class TestAggregateFunction extends AggregateFunction<String, TestAggregateFunction.TestAggregateAcc> {

    public void accumulate(TestAggregateAcc acc, String param1, String param2, String param3) {
        acc.test = param1 + param2 + param3;
    }

    @Override
    public String getValue(TestAggregateAcc accumulator) {
        return accumulator.test;
    }

    @Override
    public TestAggregateAcc createAccumulator() {
        return new TestAggregateAcc();
    }

    public static class TestAggregateAcc {
        public String test;
    }
}

public class TestAggregateFunction extends AggregateFunction<String, TestAggregateFunction.TestAggregateAcc> {

    public void accumulate(TestAggregateAcc acc, String param1, String param2, String param3) {
        acc.test = param1 + param2 + param3;
    }

    @Override
    public String getValue(TestAggregateAcc accumulator) {
        return accumulator.test;
    }

    @Override
    public TestAggregateAcc createAccumulator() {
        return new TestAggregateAcc();
    }

    public static class TestAggregateAcc {
        public String test;
    }
}

  String sql = "INSERT INTO dwd_hudi_users " +
                "SELECT " +
                "   id ," +
                "   name ," +
                "   test_aggregate(concat_ws('_', name, email), address, 'test')," +
                "   birthday ," +
                "   ts ," +
                "   DATE_FORMAT(birthday, 'yyyyMMdd') " +
                "FROM" +
                "   ods_mysql_user_detail group by id, name, birthday, ts ";

context.execute("CREATE TABLE IF NOT EXISTS ods_mysql_user_detail (" +
                "       id                  BIGINT PRIMARY KEY NOT ENFORCED ," +
                "       name                STRING                          ," +
                "       birthday            TIMESTAMP(3)                    ," +
                "       ts                  TIMESTAMP(3)                    ," +
                "       email               STRING                          ," +
                "       address             STRING                          ," +
                "       proc_time as proctime()                              " +
                ") WITH ( " +
                "       'connector' = 'mysql-cdc'            ," +
                "       'hostname'  = '127.0.0.1'       ," +
                "       'port'      = '3306'                 ," +
                "       'username'  = 'root'                 ," +
                "       'password'  = 'xxx'          ," +
                "       'server-time-zone' = 'Asia/Shanghai' ," +
                "       'database-name' = 'demo'             ," +
                "       'table-name'    = 'users' " +
                ")");