Skip to main content
Version: 0.6.0-incubating

Apache Gravitino Flink connector

Overview

The Apache Gravitino Flink connector implements the Catalog Store to manage the catalogs under Gravitino. This capability allows users to perform federation queries, accessing data from various catalogs through a unified interface and consistent access control.

Capabilities

  1. Supports Hive catalog
  2. Supports most DDL and DML SQLs.

Requirement

  • Flink 1.18
  • Scala 2.12
  • JDK 8 or 11 or 17

How to use it

  1. Build or download the Gravitino flink connector runtime jar, and place it to the classpath of Flink.
  2. Configure the Flink configuration to use the Gravitino flink connector.
PropertyTypeDefault ValueDescriptionRequiredSince Version
table.catalog-store.kindstringgeneric_in_memoryThe Catalog Store name, it should set to gravitino.Yes0.6.0
table.catalog-store.gravitino.gravitino.metalakestring(none)The metalake name that flink connector used to request to Gravitino.Yes0.6.0
table.catalog-store.gravitino.gravitino.uristring(none)The uri of Gravitino server address.Yes0.6.0

Set the flink configuration in flink-conf.yaml.

table.catalog-store.kind=gravitino
table.catalog-store.gravitino.gravitino.metalake=test
table.catalog-store.gravitino.gravitino.uri=http://localhost:8080

Or you can set the flink configuration in the TableEnvironment.

final Configuration configuration = new Configuration();
configuration.setString("table.catalog-store.kind", "gravitino");
configuration.setString("table.catalog-store.gravitino.gravitino.metalake", "test");
configuration.setString("table.catalog-store.gravitino.gravitino.uri", "http://localhost:8080");
EnvironmentSettings.Builder builder = EnvironmentSettings.newInstance().withConfiguration(configuration);
TableEnvironment tableEnv = TableEnvironment.create(builder.inBatchMode().build());
  1. Execute the Flink SQL query.

Suppose there is only one hive catalog with the name hive in the metalake test.

// use hive catalog
USE hive;
CREATE DATABASE db;
USE db;
CREATE TABLE hive_students (id INT, name STRING);
INSERT INTO hive_students VALUES (1, 'Alice'), (2, 'Bob');
SELECT * FROM hive_students;

Datatype mapping

Gravitino flink connector support the following datatype mapping between Flink and Gravitino.

Flink TypeGravitino TypeSince Version
arrayarray0.6.0
bigintlong0.6.0
binaryfixed0.6.0
booleanboolean0.6.0
charchar0.6.0
datedate0.6.0
decimaldecimal0.6.0
doubledouble0.6.0
floatfloat0.6.0
integerinteger0.6.0
mapmap0.6.0
nullnull0.6.0
rowstruct0.6.0
smallintshort0.6.0
timetime0.6.0
timestamptimestamp without time zone0.6.0
timestamp without time zonetimestamp without time zone0.6.0
timestamp with time zonetimestamp with time zone0.6.0
timestamp with local time zonetimestamp with time zone0.6.0
timestamp_ltztimestamp with time zone0.6.0
tinyintbyte0.6.0
varbinarybinary0.6.0
varcharstring0.6.0