0
点赞
收藏
分享

微信扫一扫

sap提示框function：以及与客户端文件系统的交互

非凡兔 2023-11-12 阅读 38
flink大数据Catalog元数据

Catalog

Catalog

概述

Catalog分类

1.GenericInMemoryCatalog

2.JdbcCatalog

3.HiveCatalog

4.用户自定义Catalog

GenericInMemoryCatalog

JdbcCatalog

这里以JdbcCatalog-MySQL使用为例。

下载JAR包及使用

下载：flink-connector-jdbc

下载：mysql-connector-j

上传JAR包到flink/lib下

cp ./flink-connector-jdbc-3.1.0-1.17.jar /usr/local/program/flink/lib

cp ./mysql-connector-j-8.0.33.jar /usr/local/program/flink/lib

重启操作

重启flink集群和sql-client

bin/start-cluster.sh

bin/sql-client.sh

创建Catalog

JdbcCatalog支持以下选项:

name:必需，Catalog名称

default-database:连接到的默认数据库

username: Postgres/MySQL帐户的用户名

password:帐号密码

base-url:数据库的jdbc url(不含数据库名)
	Postgres Catalog：是"jdbc:postgresql://<ip>:<端口>"
	MySQL Catalog：是"jdbc: mysql://<ip>:<端口>"

CREATE CATALOG jdbc_catalog WITH(
    'type' = 'jdbc',
    'default-database' = 'demo',
    'username' = 'root',
    'password' = '123456',
    'base-url' = 'jdbc:mysql://node01:3306'
);

查看与使用Catalog

查看Catalog

Flink SQL> show catalogs;
+-----------------+
|    catalog name |
+-----------------+
| default_catalog |
|    jdbc_catalog |
+-----------------+
2 rows in set

使用指定Catalog

Flink SQL> use catalog jdbc_catalog;
[INFO] Execute statement succeed.

查看当前的CATALOG

Flink SQL> SHOW CURRENT CATALOG;
+----------------------+
| current catalog name |
+----------------------+
|         jdbc_catalog |
+----------------------+
1 row in set

操作数据库表

Flink SQL> show current database;
+-----------------------+
| current database name |
+-----------------------+
|                  demo |
+-----------------------+
1 row in set


Flink SQL> show tables;
+------------+
| table name |
+------------+
|    tb_user |
+------------+
1 row in set

Flink SQL> select * from tb_user;
[INFO] Result retrieval cancelled.

Flink SQL> insert into tb_user values(0,'java',20);
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 9d78ec378ad635d291bd730ba86245d8

自动初始化catalog

进入SQL客户端自动初始化catalo，创建vim sql-client-init.sql初始化脚本

SET sql-client.execution.result-mode = 'tableau';

CREATE CATALOG jdbc_catalog WITH(
    'type' = 'jdbc',
    'default-database' = 'demo',
    'username' = 'root',
    'password' = '123456',
    'base-url' = 'jdbc:mysql://node01:3306'
);

use catalog jdbc_catalog;

进入客户端时指定初始化文件

bin/sql-client.sh  -i ./sql-client-init.sql

再查看catalog

Flink SQL> show catalogs;
+-----------------+
|    catalog name |
+-----------------+
| default_catalog |
|    jdbc_catalog |
+-----------------+
2 rows in set

HiveCatalog

HiveCatalog有两个用途：

单纯作为 Flink元数据的持久化存储

作为读写现有Hive元数据的接口

下载JAR包及使用

下载：flink-sql-connector-hive

下载：mysql-connector-j

上传jar包到flink的lib

cp ./flink-sql-connector-hive-2.3.9_2.12-1.17.0.jar /usr/local/program/flink/lib/

cp ./mysql-connector-j-8.0.33.jar /usr/local/program/flink/lib

重启操作

重启flink集群和sql-client

bin/start-cluster.sh

bin/sql-client.sh

hive metastore服务

启动外置的hive metastore服务

  <property>
    <name>hive.metastore.uris</name>
    <value>thrift://node01:9083</value>
  </property>

# 前台运行
hive --service metastore

# 后台运行
hive --service metastore &

创建Catalog

配置项必需默认值类型说明
typeYes(none)StringCatalog类型，创建HiveCatalog时必须设置为’hive’
nameYes(none)StringCatalog的唯一名称
hive-conf-dirNo(none)String包含hive -site.xml的目录,需要Hadoop文件系统支持。如果没指定hdfs协议，则认为是本地文件系统。如果不指定该选项，则在类路径中搜索hive-site.xml
default-databaseNodefaultStringHive Catalog使用的默认数据库
hive-versionNo(none)StringHiveCatalog能够自动检测正在使用的Hive版本。建议不要指定Hive版本，除非自动检测失败
hadoop-conf-dirNo(none)StringHadoop conf目录的路径。只支持本地文件系统路径。设置Hadoop conf的推荐方法是通过HADOOP_CONF_DIR环境变量。只有当环境变量不适合你时才使用该选项，例如，如果你想分别配置每个HiveCatalog
CREATE CATALOG myhive WITH (
    'type' = 'hive',
    'default-database' = 'default',
    'hive-conf-dir' = '/usr/local/program/hive/conf'
);

查看与使用Catalog

查看Catalog

Flink SQL> SHOW CATALOGS;
+-----------------+
|    catalog name |
+-----------------+
| default_catalog |
|          myhive |
+-----------------+
2 rows in set


--查看当前的CATALOG
SHOW CURRENT CATALOG;

使用指定Catalog

Flink SQL> use catalog myhive;
[INFO] Execute statement succeed.

Flink与Hive中操作

Flink中查看

Flink SQL> SHOW DATABASES;
+---------------+
| database name |
+---------------+
|       default |
+---------------+
1 row in set

操作Hive

# 创建数据库demo
hive (default)> create database demo;

# 切换数据库
hive (default)> use demo;

# 创建表tb_user
hive (demo)> create table tb_user(id int,name string, age int);

# 插入数据
hive (demo)> insert into tb_user values(1,"test",22);

Flink中再次查看

Flink SQL> SHOW DATABASES;
+---------------+
| database name |
+---------------+
|       default |
|          demo |
+---------------+
2 rows in set

Flink SQL> use demo;
[INFO] Execute statement succeed.

Flink SQL> show tables;
+------------+
| table name |
+------------+
|    tb_user |
+------------+


Flink SQL> SET 'sql-client.execution.result-mode' = 'tableau';
[INFO] Execute statement succeed.

Flink SQL> select * from tb_user;2023-07-09 21:58:25,620 INFO  org.apache.hadoop.mapred.FileInputFormat                     [] - Total input files to process : 1

+----+-------------+--------------------------------+-------------+
| op |          id |                           name |         age |
+----+-------------+--------------------------------+-------------+
| +I |           1 |                           test |          22 |
+----+-------------+--------------------------------+-------------+
Received a total of 1 row

在Flink中插入

Flink SQL> insert into tb_user values(2,'flink',22);
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 9fe32af97cfb9e507ce84263cae65d23

Flink SQL> select * from tb_user;2023-07-09 22:05:47,521 INFO  org.apache.hadoop.mapred.FileInputFormat                     [] - Total input files to process : 2

+----+-------------+--------------------------------+-------------+
| op |          id |                           name |         age |
+----+-------------+--------------------------------+-------------+
| +I |           1 |                           test |          22 |
| +I |           2 |                          flink |          22 |
+----+-------------+--------------------------------+-------------+
Received a total of 2 rows

Hive中查询

hive (demo)> select * from tb_user;

自动初始化catalog

进入SQL客户端自动初始化catalog，创建vim sql-client-init.sql初始化脚本

SET sql-client.execution.result-mode = 'tableau';

CREATE CATALOG myhive WITH (
    'type' = 'hive',
    'default-database' = 'default',
    'hive-conf-dir' = '/usr/local/program/hive/conf'
);

use catalog myhive ;

进入客户端时指定初始化文件

bin/sql-client.sh  -i ./sql-client-init.sql

可以发现数据信息任然存在

Flink SQL> use catalog myhive;
[INFO] Execute statement succeed.

Flink SQL> show databases;
+---------------+
| database name |
+---------------+
|       default |
|          demo |
+---------------+
2 rows in set

用户自定义Catalog

实现Catalog

public class CustomCatalog implements Catalog {

    public CustomCatalog(String catalogName, String defaultDatabase) {
        
    }


    @Override
    public void open() {
        // 实现 Catalog 打开的逻辑
    }

    @Override
    public void close() {
        // 实现 Catalog 关闭的逻辑
    }

    @Override
    public List<String> listDatabases() {
        // 实现获取数据库列表的逻辑
        return null;
    }

    @Override
    public CatalogDatabase getDatabase(String databaseName) {
        // 实现获取指定数据库的逻辑
        return null;
    }

    @Override
    public boolean databaseExists(String databaseName) {
        // 实现检查数据库是否存在的逻辑
        return false;
    }

    @Override
    public void createDatabase(String name, CatalogDatabase database, boolean ignoreIfExists) {
        // 实现创建数据库的逻辑
    }

    @Override
    public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade) {
        // 实现删除数据库的逻辑
    }

    @Override
    public List<String> listTables(String databaseName) {
        // 实现获取数据库中表的列表的逻辑
        return null;
    }

    @Override
    public CatalogBaseTable getTable(ObjectPath tablePath) {
        // 实现获取指定表的逻辑
        return null;
    }

    @Override
    public boolean tableExists(ObjectPath tablePath) {
        // 实现检查表是否存在的逻辑
        return false;
    }

    @Override
    public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) {
        // 实现创建表的逻辑
    }

    @Override
    public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists) {
        // 实现删除表的逻辑
    }

    @Override
    public List<String> listFunctions(String dbName) {
        // 实现获取数据库中函数的逻辑
        return null;
    }

    // 其他方法的实现
}

使用Catalog

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 注册自定义 Catalog
        tableEnv.registerCatalog("my_catalog", new CustomCatalog("my_catalog", "default"));

        // 使用自定义 Catalog
        tableEnv.useCatalog("my_catalog");

        // 执行 SQL 查询或 Table API 操作
        tableEnv.sqlQuery("SELECT * FROM my_table").execute().print();
    }

Catalog API

数据库操作

    public static void main(String[] args) throws Exception {
        // 创建一个基于内存的Catalog实例
        GenericInMemoryCatalog catalog = new GenericInMemoryCatalog("myCatalog");
        catalog.open();


        // 创建数据库
        Map<String, String> properties = new HashMap<>();
        properties.put("key", "value");
        CatalogDatabase database = new CatalogDatabaseImpl(properties, "create comment");
        catalog.createDatabase("mydb", database, false);

        // 列出Catalog中的所有数据库
        System.out.println("列出Catalog中的所有数据库 = " + catalog.listDatabases());

        // 获取数据库
        CatalogDatabase createDb = catalog.getDatabase("mydb");
        System.out.println("获取数据库,comment =  " + createDb.getComment() + " ,properties = " + createDb.getProperties());

        // 修改数据库
        Map<String, String> properties2 = new HashMap<>();
        properties2.put("key", "value1");
        catalog.alterDatabase("mydb", new CatalogDatabaseImpl(properties2, "alter comment"), false);

        // 获取数据库
        CatalogDatabase alterDb = catalog.getDatabase("mydb");
        System.out.println("获取数据库,comment =  " + alterDb.getComment() + " ,properties = " + alterDb.getProperties());

        // 检查数据库是否存在
        System.out.println("检查数据库是否存在 = " + catalog.databaseExists("mydb"));

        // 删除数据库
        catalog.dropDatabase("mydb", false);

        // 关闭 Catalog
        catalog.close();
    }

列出Catalog中的所有数据库 = [default, mydb]
获取数据库,comment =  create comment ,properties = {key=value}
获取数据库,comment =  alter comment ,properties = {key=value1}
检查数据库是否存在 = true

表操作

// 创建表
catalog.createTable(new ObjectPath("mydb", "mytable"), new CatalogTableImpl(...), false);

// 删除表
catalog.dropTable(new ObjectPath("mydb", "mytable"), false);

// 修改表
catalog.alterTable(new ObjectPath("mydb", "mytable"), new CatalogTableImpl(...), false);

// 重命名表
catalog.renameTable(new ObjectPath("mydb", "mytable"), "my_new_table");

// 获取表
catalog.getTable("mytable");

// 检查表是否存在
catalog.tableExists("mytable");

// 列出数据库中的所有表
catalog.listTables("mydb");

视图操作

// 创建视图
catalog.createTable(new ObjectPath("mydb", "myview"), new CatalogViewImpl(...), false);

// 删除视图
catalog.dropTable(new ObjectPath("mydb", "myview"), false);

// 修改视图
catalog.alterTable(new ObjectPath("mydb", "mytable"), new CatalogViewImpl(...), false);

// 重命名视图
catalog.renameTable(new ObjectPath("mydb", "myview"), "my_new_view", false);

// 获取视图
catalog.getTable("myview");

// 检查视图是否存在
catalog.tableExists("mytable");

// 列出数据库中的所有视图
catalog.listViews("mydb");

分区操作

// 创建分区
catalog.createPartition(
    new ObjectPath("mydb", "mytable"),
    new CatalogPartitionSpec(...),
    new CatalogPartitionImpl(...),
    false);

// 删除分区
catalog.dropPartition(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...), false);

// 修改分区
catalog.alterPartition(
    new ObjectPath("mydb", "mytable"),
    new CatalogPartitionSpec(...),
    new CatalogPartitionImpl(...),
    false);

// 获取分区
catalog.getPartition(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));

// 检查分区是否存在
catalog.partitionExists(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));

// 列出表的所有分区
catalog.listPartitions(new ObjectPath("mydb", "mytable"));

// 根据给定的分区规范列出表的分区
catalog.listPartitions(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));

// 根据表达式过滤器列出表的分区
catalog.listPartitions(new ObjectPath("mydb", "mytable"), Arrays.asList(epr1, ...));

函数操作

// 创建函数
catalog.createFunction(new ObjectPath("mydb", "myfunc"), new CatalogFunctionImpl(...), false);

// 删除函数
catalog.dropFunction(new ObjectPath("mydb", "myfunc"), false);

// 修改函数
catalog.alterFunction(new ObjectPath("mydb", "myfunc"), new CatalogFunctionImpl(...), false);

// 获取函数
catalog.getFunction("myfunc");

// 检查函数是否存在
catalog.functionExists("myfunc");

// 列出数据库中的所有函数
catalog.listFunctions("mydb");
举报

相关推荐

Android对话框合集 以及提示框

androididexmlRedis数据库

iOS 隐藏系统音量提示框

ideicoios虚拟化云计算

Android实战简易教程-第五十六枪（模拟美团客户端进度提示框）

Android美团动画androidxmlLinux系统/运维

Qt服务器端与客户端交互

socketQt客户端与服务器端交互

客户端服务端交互概述

HTTPWebHTML代码人生yyds干货盘点

客户端服务端交互实现

网络udp网络协议

网络文件系统nfs服务端配置客户端权限时的demo例子

文件系统（磁盘文件系统）Linux分布式系统客户端服务端OpenStack云计算

Eureka Client客户端如何与注册中心交互？

Spring Cloud饥人谷技术博客微服务架构和实践SpringF...

Android 常用的提示框Dialog

android dialogidegithubcijQuery前端开发

Android中的弹出提示框

移动开发javaAndroidf5提示框Html/CSS前端开发
0 条评论