window上运行Flink 读取hive catalog 代码实操

不念不忘少年蓝@ 2023-02-13 08:52 8阅读 0赞

一，导入依赖，主要是hive 跟  hadoop

>     
>     <dependency>
>         <groupId>org.apache.flink</groupId>
>         <artifactId>flink-connector-hive_2.11</artifactId>
>         <version>1.10.0</version>
>         
>     </dependency>
>     
>     <dependency>
>         <groupId>org.apache.hive</groupId>
>         <artifactId>hive-exec</artifactId>
>         <version>1.1.0</version>
>         
>     </dependency>
>     
>     
>     <dependency>
>         <groupId>org.apache.hadoop</groupId>
>         <artifactId>hadoop-common</artifactId>
>         <version>2.6.0-cdh5.16.1</version>
>     </dependency>
>     
>     <dependency>
>         <groupId>org.apache.hadoop</groupId>
>         <artifactId>hadoop-hdfs</artifactId>
>         <version>2.6.0-cdh5.16.1</version>
>     </dependency>
>     
>     <dependency>
>         <groupId>org.apache.hadoop</groupId>
>         <artifactId>hadoop-client</artifactId>
>         <version>2.6.0-cdh5.16.1</version>
>     </dependency>

二，在window环境下最好有hadoop的环境变量，这个请自己百度。

三，就是搞一个hive-site.xml配置文件了。 参考官网 主要是几个指标

1）一定要看自己的hive版本

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

[https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/hive/][https_ci.apache.org_projects_flink_flink-docs-release-1.10_dev_table_hive]

2）配置文件模板

<configuration>
       <property>
          <name>javax.jdo.option.ConnectionURL</name>
          <value>jdbc:mysql://localhost/metastore?createDatabaseIfNotExist=true</value>
          <description>metadata is stored in a MySQL server</description>
       </property>
    
       <property>
          <name>javax.jdo.option.ConnectionDriverName</name>
          <value>com.mysql.jdbc.Driver</value>
          <description>MySQL JDBC driver class</description>
       </property>
    
       <property>
          <name>javax.jdo.option.ConnectionUserName</name>
          <value>...</value>
          <description>user name for connecting to mysql server</description>
       </property>
    
       <property>
          <name>javax.jdo.option.ConnectionPassword</name>
          <value>...</value>
          <description>password for connecting to mysql server</description>
       </property>
    
       <property>
           <name>hive.metastore.uris</name>
           <value>thrift://localhost:9083</value>
           <description>IP address (or fully-qualified domain name) and port of the metastore host</description>
       </property>
    
       <property>
           <name>hive.metastore.schema.verification</name>
           <value>true</value>
       </property>
    
    </configuration>

三，最后就是代码了 很简单的读取代码：

>  
> 
>     package flink_sql
>     
>     
>     import org.apache.flink.streaming.api.TimeCharacteristic
>     import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
>     import org.apache.flink.table.api.EnvironmentSettings
>     import org.apache.flink.table.api.scala.StreamTableEnvironment
>     import org.apache.flink.table.catalog.hive.HiveCatalog
>     import org.apache.flink.table.api.Table
>     
>     object Sql_client_scala {
>       def main(args: Array[String]): Unit = {
>         val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
>         streamEnv.setParallelism(5)
>         streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
>     
>         val tableEnvSettings = EnvironmentSettings.newInstance()
>           .useBlinkPlanner()
>           .inStreamingMode()
>           .build()
>         val tableEnv = StreamTableEnvironment.create(streamEnv, tableEnvSettings)
>     
>         val catalog = new HiveCatalog(
>           "flink", // catalog name
>           "default", // default database
>           "G:\\Flink SQL开发文件", // Hive config (hive-site.xml) directory
>           "1.1.0" // Hive version
>         )
>     
>         //todo 注册这个catalog
>         tableEnv.registerCatalog("flink", catalog)
>     
>         //todo 使用这个catalog，这个表在内存
>         tableEnv.useCatalog("flink")
>     
>     //    val createDbSql1 = "CREATE DATABASE IF NOT EXISTS flink"
>     //    tableEnv.sqlUpdate(createDbSql1)
>     
>         //todo 测试创建表
>     //    val createtable = "CREATE TABLE flink_test (id STRING, product STRING, amount INT)"
>     //    tableEnv.sqlUpdate(createtable)
>     
>         //todo 存在哪些库
>         val aa: Array[String] = tableEnv.listCatalogs()
>         print(aa.toList)
>     
>         //todo 存在的表
>         val tables = tableEnv.listTables()
>         println(tables.toList)
>     
>         /**
>           * todo 从kafka读取数据构建表
>           */
>     
>         //todo 查询sql
>         val createDbSql = "SELECT code ,total_emp FROM  sample_07"
>         val rTable = tableEnv.sqlQuery(createDbSql)
>          println(rTable.printSchema())
>     
>         import org.apache.flink.api.scala._
>     
>         //todo 查询数据
>         val rsss: DataStream[(String, Int)] =   tableEnv.toAppendStream[(String, Int)](rTable)
>         rsss.print()
>         streamEnv.execute()
>          //    val  sql ="CREATE TABLE rtdw.ods. (name String, age Int) WITH (\n" +
>         //      "   'connector.type' = 'kafka',\n" +
>         //      "   'connector.version' = 'universal',\n" +
>         //      "   'connector.topic' = 'test1',\n" +
>         //      "   'connector.properties.zookeeper.connect' = 'dev-ct6-dc-worker01:2181,dev-ct6-dc-worker02:2181,dev-ct6-dc-worker03:2181',\n" +
>         //      "   'connector.properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',\n" +
>         //      "   'format.type' = 'json',\n" +
>         //      "   'update-mode' = 'append'\n" +
>         //      ");"
>         //    tableEnv.sqlUpdate(createDbSql)
>       }
>     
>     }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200528103047232.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz,size_16,color_FFFFFF,t_70
[https_ci.apache.org_projects_flink_flink-docs-release-1.10_dev_table_hive]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/hive/