• 使用FlinkCatalog将kafka的数据写入hive

    1. package com.atguigu.flink.test_hk;
    2.  
    3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    4. import org.apache.flink.table.api.EnvironmentSettings;
    5. import org.apache.flink.table.api.SqlDialect;
    6. import org.apache.flink.table.api.TableEnvironment;
    7. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    8. import org.apache.flink.table.catalog.hive.HiveCatalog;
    9.  
    10. public class KafkaToHive3 {
    11.  
    12.     public static void main(String[] args) {
    13.  
    14.         System.setProperty("HADOOP_USER_NAME", "atguigu");
    15.  
    16.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    17.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    18.         env.setParallelism(1);
    19.  
    20.         env.enableCheckpointing(6000);
    21.  
    22. // to use hive dialect
    23.         tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
    24.  
    25.         // 注册Hive Catalog
    26.         HiveCatalog hiveCatalog = new HiveCatalog("hive", "default", "conf");
    27.         tableEnv.registerCatalog("hive", hiveCatalog);
    28.  
    29. // 使用默认的Flink catalog创建Kafka表
    30.         tableEnv.useCatalog("hive");
    31.  
    32. // to use default dialect
    33. //        tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
    34.  
    35.         String hivesql = "CREATE TABLE  IF NOT EXISTS hive_table7 (\n" +
    36.                 "  id int\n" +
    37.                 ") ";
    38.  
    39.         tableEnv.executeSql(hivesql);
    40.  
    41.  
    42.  
    43. //        tableEnv.sqlQuery("select * from hive_table7  ").execute().print();
    44.  
    45.         // 使用默认的Flink catalog创建Kafka表
    46.         tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
    47.         tableEnv.useCatalog("default_catalog");
    48.  
    49.         String readSql = " create table t2 (" +
    50.                 " id int  " +
    51.                 ") WITH (" +
    52.                 " 'connector' = 'kafka', " +
    53.                 " 'topic' = 'topicA', " +
    54.                 " 'properties.bootstrap.servers' = 'hadoop102:9092', " +
    55.                 " 'properties.group.id' = 'flin_"+System.currentTimeMillis()+"', " +
    56.                 " 'scan.startup.mode' = 'latest-offset', " +
    57.                 " 'value.format' = 'csv' " +
    58.                 ")" ;
    59.         tableEnv.executeSql(readSql);
    60.  
    61. //        tableEnv.sqlQuery("select * from t2  ").execute().print();
    62.  
    63.         tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
    64.         tableEnv.useCatalog("hive");
    65.  
    66.         tableEnv.executeSql("INSERT INTO hive_table7 select id from default_catalog.default_database.t2");
    67.  
    68.  
    69.     }
    70.  
    71. }
    1. "1.0" encoding="UTF-8"?>
    2. <project xmlns="http://maven.apache.org/POM/4.0.0"
    3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    5.     <parent>
    6.         <artifactId>BigData230201artifactId>
    7.         <groupId>com.atguigugroupId>
    8.         <version>1.0-SNAPSHOTversion>
    9.     parent>
    10.     <modelVersion>4.0.0modelVersion>
    11.  
    12.     <artifactId>FlinkartifactId>
    13.  
    14.     <properties>
    15.         <maven.compiler.source>8maven.compiler.source>
    16.         <maven.compiler.target>8maven.compiler.target>
    17.         <project.build.sourceEncoding>UTF-8project.build.sourceEncoding>
    18.         <flink.version>1.17.0flink.version>
    19.         <scala.version>2.12scala.version>
    20.     properties>
    21.  
    22.     
    23.  
    24.  
    25.     <dependencies>
    26.         <dependency>
    27.             <groupId>org.apache.flinkgroupId>
    28.             <artifactId>flink-streaming-javaartifactId>
    29.             <version>${flink.version}version>
    30.             <scope>providedscope>
    31.         dependency>
    32.  
    33.         <dependency>
    34.             <groupId>org.apache.flinkgroupId>
    35.             <artifactId>flink-clientsartifactId>
    36.             <version>${flink.version}version>
    37.             <scope>providedscope>
    38.         dependency>
    39.         <dependency>
    40.             <groupId>org.slf4jgroupId>
    41.             <artifactId>slf4j-log4j12artifactId>
    42.             <version>1.7.25version>
    43.             <scope>providedscope>
    44.         dependency>
    45.  
    46.         
    47.         <dependency>
    48.             <groupId>org.projectlombokgroupId>
    49.             <artifactId>lombokartifactId>
    50.             <version>1.18.26version>
    51.         dependency>
    52.  
    53.         
    54.         <dependency>
    55.             <groupId>org.apache.flinkgroupId>
    56.             <artifactId>flink-runtime-webartifactId>
    57.             <version>${flink.version}version>
    58.             <scope>providedscope>
    59.         dependency>
    60.  
    61.         
    62.         <dependency>
    63.             <groupId>org.apache.flinkgroupId>
    64.             <artifactId>flink-connector-datagenartifactId>
    65.             <version>${flink.version}version>
    66.             <scope>providedscope>
    67.         dependency>
    68.  
    69.  
    70.         
    71.         <dependency>
    72.             <groupId>org.apache.flinkgroupId>
    73.             <artifactId>flink-connector-filesartifactId>
    74.             <version>${flink.version}version>
    75.             <scope>providedscope>
    76.         dependency>
    77.  
    78.         
    79.         <dependency>
    80.             <groupId>org.apache.hadoopgroupId>
    81.             <artifactId>hadoop-clientartifactId>
    82.             <version>3.3.4version>
    83.             <scope>providedscope>
    84.         dependency>
    85.  
    86.         
    87.         <dependency>
    88.             <groupId>org.apache.flinkgroupId>
    89.             <artifactId>flink-connector-kafkaartifactId>
    90.             <version>${flink.version}version>
    91.             <scope>providedscope>
    92.         dependency>
    93.  
    94.         
    95.         <dependency>
    96.             <groupId>com.alibabagroupId>
    97.             <artifactId>fastjsonartifactId>
    98.             <version>1.2.83version>
    99.             <scope>providedscope>
    100.         dependency>
    101.  
    102.         
    103.         <dependency>
    104.             <groupId>org.apache.flinkgroupId>
    105.             <artifactId>flink-connector-jdbcartifactId>
    106.             <version>3.1.0-1.17version>
    107.             <scope>providedscope>
    108.         dependency>
    109.  
    110.         
    111.         <dependency>
    112.             <groupId>com.mysqlgroupId>
    113.             <artifactId>mysql-connector-jartifactId>
    114.             <version>8.0.32version>
    115.             <scope>providedscope>
    116.         dependency>
    117.  
    118.         
    119.         <dependency>
    120.             <groupId>org.apache.flinkgroupId>
    121.             <artifactId>flink-statebackend-rocksdbartifactId>
    122.             <version>${flink.version}version>
    123.             <scope>providedscope>
    124.         dependency>
    125.  
    126.         
    127.         <dependency>
    128.             <groupId>org.apache.flinkgroupId>
    129.             <artifactId>flink-statebackend-changelogartifactId>
    130.             <version>${flink.version}version>
    131.             <scope>providedscope>
    132.         dependency>
    133.  
    134.         
    135.         <dependency>
    136.             <groupId>org.apache.flinkgroupId>
    137.             <artifactId>flink-table-api-java-uberartifactId>
    138.             <version>${flink.version}version>
    139.         dependency>
    140.  
    141.         <dependency>
    142.             <groupId>org.apache.flinkgroupId>
    143.             <artifactId>flink-table-runtimeartifactId>
    144.             <version>${flink.version}version>
    145.         dependency>
    146.  
    147.  
    153.  
    154.  
    155.         <dependency>
    156.             <groupId>org.apache.flinkgroupId>
    157.             <artifactId>flink-table-planner_${scala.version}artifactId>
    158.             <version>${flink.version}version>
    159.         dependency>
    160.  
    161.         
    162.         <dependency>
    163.             <groupId>org.apache.flinkgroupId>
    164.             <artifactId>flink-csvartifactId>
    165.             <version>${flink.version}version>
    166.         dependency>
    167.  
    168.         
    169.         <dependency>
    170.             <groupId>org.apache.flinkgroupId>
    171.             <artifactId>flink-jsonartifactId>
    172.             <version>${flink.version}version>
    173.         dependency>
    174.  
    175.         
    176.         <dependency>
    177.             <groupId>org.apache.flinkgroupId>
    178.             <artifactId>flink-connector-hive_2.12artifactId>
    179.             <version>${flink.version}version>
    180.             <scope>providedscope>
    181.         dependency>
    182.  
    183.         <dependency>
    184.             <groupId>org.apache.hivegroupId>
    185.             <artifactId>hive-execartifactId>
    186.             <version>3.1.3version>
    187.             <scope>providedscope>
    188.         dependency>
    189.  
    190.  
    191.  
    192.  
    193.  
    194.  
    195.     dependencies>
    196.  
    197.     <build>
    198.         <plugins>
    199.             <plugin>
    200.                 <groupId>org.apache.maven.pluginsgroupId>
    201.                 <artifactId>maven-shade-pluginartifactId>
    202.                 <version>3.2.4version>
    203.                 <executions>
    204.                     <execution>
    205.                         <phase>packagephase>
    206.                         <goals>
    207.                             <goal>shadegoal>
    208.                         goals>
    209.                         <configuration>
    210.                             <artifactSet>
    211.                                 <excludes>
    212.                                     <exclude>com.google.code.findbugs:jsr305exclude>
    213.                                     <exclude>org.slf4j:*exclude>
    214.                                     <exclude>log4j:*exclude>
    215.                                 excludes>
    216.                             artifactSet>
    217.                             <filters>
    218.                                 <filter>
    219.                                     
    220.                                     <artifact>*:*artifact>
    221.                                     <excludes>
    222.                                         <exclude>META-INF/*.SFexclude>
    223.                                         <exclude>META-INF/*.DSAexclude>
    224.                                         <exclude>META-INF/*.RSAexclude>
    225.                                     excludes>
    226.                                 filter>
    227.                             filters>
    228.                             <transformers combine.children="append">
    229.                                 <transformer
    230.                                         implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer">
    231.                                 transformer>
    232.                             transformers>
    233.                         configuration>
    234.                     execution>
    235.                 executions>
    236.             plugin>
    237.         plugins>
    238.     build>
    239.  
    240.  
    241.  
    242.  
    243. project>

    hive-site.xml

    1. "1.0"?>
    2. "text/xsl" href="configuration.xsl"?>
    3. <configuration>
    4.     
    5.     <property>
    6.         <name>javax.jdo.option.ConnectionURLname>
    7.         <value>jdbc:mysql://hadoop102:3306/metastore?useSSL=false&useUnicode=true&characterEncoding=UTF-8&allowPublicKeyRetrieval=truevalue>
    8.     property>
    9.  
    10.     
    11.     <property>
    12.         <name>javax.jdo.option.ConnectionDriverNamename>
    13.         <value>com.mysql.cj.jdbc.Drivervalue>
    14.     property>
    15.  
    16.     
    17.     <property>
    18.         <name>javax.jdo.option.ConnectionUserNamename>
    19.         <value>rootvalue>
    20.     property>
    21.  
    22.     
    23.     <property>
    24.         <name>javax.jdo.option.ConnectionPasswordname>
    25.         <value>000000value>
    26.     property>
    27.  
    28.     <property>
    29.         <name>hive.metastore.warehouse.dirname>
    30.         <value>/user/hive/warehousevalue>
    31.     property>
    32.  
    33.     <property>
    34.         <name>hive.metastore.schema.verificationname>
    35.         <value>falsevalue>
    36.     property>
    37.  
    38.     <property>
    39.     <name>hive.server2.thrift.portname>
    40.     <value>10000value>
    41.     property>
    42.  
    43.     <property>
    44.         <name>hive.server2.thrift.bind.hostname>
    45.         <value>hadoop102value>
    46.     property>
    47.  
    48.     <property>
    49.         <name>hive.metastore.event.db.notification.api.authname>
    50.         <value>falsevalue>
    51.     property>
    52.     
    53.     <property>
    54.         <name>hive.cli.print.headername>
    55.         <value>truevalue>
    56.     property>
    57.  
    58.     <property>
    59.         <name>hive.cli.print.current.dbname>
    60.         <value>truevalue>
    61.     property>
    63. <property>
    64.     <name>spark.yarn.jarsname>
    65.     <value>hdfs://hadoop102:8020/spark-jars/*value>
    66. property>
    67.   
    69. <property>
    70.     <name>hive.execution.enginename>
    71.     <value>sparkvalue>
    72. property>
    73.  
    74. <property>
    75. 	<name>metastore.storage.schema.reader.implname>
    76. 	<value>org.apache.hadoop.hive.metastore.SerDeStorageSchemaReadervalue>
    77. property>
    78.  
    79. <property>
    80.     <name>hive.metastore.urisname>
    81.     <value>thrift://hadoop102:9083value>
    82. property>
    83.  
    84.  
    85.  
    86. configuration>
    87.

  • 相关阅读:
    商城小程序？
    【软件测试】我们测试人搭上元宇宙的列车，测试一直在进军......
    GaussDB CN服务异常实例分析
    储能领域 / 通讯协议 / 技术栈 等专有名字集锦——主要收集一些储能领域的专有名词，以及相关的名词
    企望制造ERP存在远程命令执行漏洞 附POC
    17-Linux进程管理
    系列二、类装载器ClassLoader
    不看后悔！第一本全面详解Transformer的综合性书籍！284页pdf下载
    uniapp使用sqlite
    记一次在amd架构打包arm64架构的镜像的试错经历
  • 原文地址：https://blog.csdn.net/qq_40382400/article/details/133861670