flink chunjun流处理

🧙 Questions

• hdfs输入hive输出

☄️ Ideas

hdfs输入hive输出

从hdfs上获取数据，对数据进行封装，再返回到新的hive表

{
  "job": {
    "content": [
      {
       "reader": {
          "name" : "hdfsreader", 
          "table": {
            "tableName": "users"
          },
          "parameter" : {
            "path" : "/user/hive/warehouse/users/",
            "defaultFS" : "hdfs://isxcode:9000",
            "fileType" : "text",
            "fieldDelimiter":",",
            "encoding":"UTF-8",
            "column": [
              {
                "name": "username",
                "type": "string"
              },
              {
                "name": "age",
                "type": "int"
              }
            ],
            "hadoopConfig":{
              "dfs.client.use.datanode.hostname":"true",
              "dfs.replication":"1",
              "fs.defaultFS": "hdfs://isxcode:9000",
              "hadoop.user.name": "ispong"
            }
          }
        },
        "transformer": {
          "transformSql": "select username, if(age>14, 100, 10) from users"
        },
        "writer": {
          "name" : "hivewriter",
          "table": {
            "tableName": "users_sink"
          },
          "parameter" : {
            "jdbcUrl" : "jdbc:hive2://isxcode:10000/default",
            "username" : "ispong",
            "password" : "",
            "fileType" : "text",
            "fieldDelimiter" : ",",
            "writeMode" : "overwrite",
            "partition": "pt",
            "partitionType": "DAY",
            "tablesColumn" : "{\"users_sink\":[{\"key\":\"username\",\"type\":\"string\"},{\"key\":\"age\",\"type\":\"int\"}]}",
            "defaultFS" : "hdfs://isxcode:9000"
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 1
      } 
    }
  }
}

提交作业

cd /opt/chunjun/bin
./chunjun-yarn-session.sh -job /home/ispong/test.json -confProp {\"yarn.application.id\":\"application_1665455628674_0002\"}

🔗 Links

• flink docs
• chunjun website
• flink 官方自带函数