hive
package com.hive;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import java.time.Duration;
public class HiveTest {
private static final String path = "hdfs_path";
public static void main(String []args) {
System.setProperty("HADOOP_USER_NAME", "work");
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
//
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.setStateBackend(new FsStateBackend(path));
EnvironmentSettings tableEnvSettings = EnvironmentSettings.newInstance()
.useBlinkPlanner()
.inStreamingMode()
.build();
StreamTableEnvironment tableEnv =
StreamTableEnvironment.create(env,tableEnvSettings);
tableEnv.getConfig().getConfiguration().set(ExecutionCheckpointingOptions.CHECKPOINTING_MODE,
CheckpointingMode.EXACTLY_ONCE);
tableEnv.getConfig().getConfiguration().set(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL,
Duration.ofSeconds(20));
String name= "myhive";
String defaultDatabase = "situation";
String hiveConfDir = "/load/data/hive/hive-conf"; // a local path
String version = "1.2.1";
HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir,
version);
tableEnv.registerCatalog("myhive", hive);
// set the HiveCatalog as the current catalog of the session
tableEnv.useCatalog("myhive");
tableEnv.executeSql("CREATE DATABASE IF NOT EXISTS situation");
tableEnv.executeSql("DROP TABLE IF EXISTS situation.source_table");
tableEnv.executeSql("CREATE TABLE situation.source_table (\n" +
"\thost STRING,\n" +
"\turl STRING,\n" +
"\tpublic_date STRING\n" +
") WITH (\n" +
"\t'connector.type' = 'kafka',\n" +
"\t'connector.version' = 'universal',\n" +
"\t'connector.startup-mode' = 'latest-offset',\n" +
"\t'connector.topic' = 'sendMessage',\n" +
"\t'connector.properties.group.id' = 'domain_testGroup',\n" +
"\t'connector.properties.zookeeper.connect' =
'127.0.0.1:2181',\n" +
"\t'connector.properties.bootstrap.servers' =
'127.0.0.1:9092',\n" +
"\t'update-mode' = 'append',\n" +
"\t'format.type' = 'json',\n" +
"\t'format.derive-schema' = 'true'\n" +
")");
tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
tableEnv.executeSql("DROP TABLE IF EXISTS situation.fs_table");
String hiveSql = "\n" +
" CREATE TABLE situation.fs_table (\n" +
" \n" +
"host STRING,\n" +
"url STRING,\n" +
"public_date STRING\n" +
" \n" +
" ) PARTITIONED BY (\n" +
"ts_date STRING,\n" +
"ts_hour STRING,\n" +
"ts_minute STRING\n" +
" ) STORED AS PARQUET\n" +
" TBLPROPERTIES (\n" +
"'sink.partition-commit.trigger' = 'process time',\n" +
"'sink.partition-commit.delay' = '1 min',\n" +
"'sink.partition-commit.policy.kind' =
'metastore,success-file',\n" +
"'partition.time-extractor.timestamp-pattern' = '$ts_date
$ts_hour:$ts_minute:00'\n" +
" )\n" +
" ";
tableEnv.executeSql(hiveSql);
tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
tableEnv.executeSql("INSERT INTO situation.fs_table SELECT host,
url,public_date," +
" DATE_FORMAT(public_date,'-MM-dd')
,DATE_FORMAT(public_date,'HH') ,DATE_FORMAT(public_date,'mm') FROM
situation.source_table");
}
}
