[#3515] feat(flink-connector): Support flink iceberg catalog #5914

sunxiaojian · 2024-12-19T07:05:30Z

What changes were proposed in this pull request?

Support flink iceberg catalog

Why are the changes needed?

Fix: #3515

Does this PR introduce any user-facing change?

no

How was this patch tested?

FlinkIcebergCatalogIT
FlinkIcebergHiveCatalogIT

sunxiaojian · 2024-12-19T09:57:27Z

@FANNG1 @coolderli PTAL

FANNG1 · 2024-12-19T11:05:45Z

Cool!, I'll review this PR, but may need some time, :)

sunxiaojian · 2024-12-19T11:28:37Z

Cool!, I'll review this PR, but may need some time, :)

ok, thanks

tengqm · 2024-12-22T04:24:42Z

docs/flink-connector/flink-catalog-iceberg.md

+The Apache Gravitino Flink connector offers the capability to read and write Iceberg tables, with the metadata managed by the Gravitino server. To enable the use of the Iceberg catalog within the Flink connector, you must download the Iceberg Flink runtime JAR to the Flink classpath.
+


Suggested change

The Apache Gravitino Flink connector offers the capability to read and write Iceberg tables, with the metadata managed by the Gravitino server. To enable the use of the Iceberg catalog within the Flink connector, you must download the Iceberg Flink runtime JAR to the Flink classpath.

The Apache Gravitino Flink connector can be used to read and write Iceberg tables, with the metadata managed by the Gravitino server.

To enable the Flink connector, you must download the Iceberg Flink runtime JAR and place it in the Flink classpath.

tengqm · 2024-12-22T04:25:13Z

docs/flink-connector/flink-catalog-iceberg.md

+- `INSERT INTO & OVERWRITE`
+- `SELECT`
+
+#### Not supported operations:


Suggested change

#### Not supported operations:

#### Operations Not Supported:

tengqm · 2024-12-22T04:26:49Z

docs/flink-connector/flink-catalog-iceberg.md

+
+## Catalog properties
+
+Gravitino Flink connector will transform the following property names defined in catalog properties to Flink Iceberg connector configuration.


Suggested change

Gravitino Flink connector will transform the following property names defined in catalog properties to Flink Iceberg connector configuration.

The Gravitino Flink connector transforms the following properties in a catalog to Flink connector configuration.

tengqm · 2024-12-22T04:28:02Z

docs/flink-connector/flink-catalog-iceberg.md

+
+### S3
+
+You need to add s3 secret to the Flink configuration using `s3.access-key-id` and `s3.secret-access-key`. Additionally, download the [Iceberg AWS bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-aws-bundle) and place it in the classpath of Flink.


Suggested change

You need to add s3 secret to the Flink configuration using `s3.access-key-id` and `s3.secret-access-key`. Additionally, download the [Iceberg AWS bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-aws-bundle) and place it in the classpath of Flink.

You need to add an S3 secret to the Flink configuration using `s3.access-key-id` and `s3.secret-access-key`.

Additionally, you need to download the [Iceberg AWS bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-aws-bundle)

and place it in the Flink classpath.

tengqm · 2024-12-22T04:28:48Z

docs/flink-connector/flink-catalog-iceberg.md

+
+### OSS
+
+You need to add OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`. Additionally, download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip) and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` in the classpath of Flink.


Suggested change

You need to add OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`. Additionally, download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip) and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` in the classpath of Flink.

You need to add an OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`.

Additionally, you need download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip),

and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` to the Flink classpath.

tengqm · 2024-12-22T04:29:49Z

docs/flink-connector/flink-catalog-iceberg.md

+
+### GCS
+
+No extra configuration is needed. Please make sure the credential file is accessible by Flink, like using `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`, and download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it to the classpath of Flink.


Suggested change

No extra configuration is needed. Please make sure the credential file is accessible by Flink, like using `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`, and download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it to the classpath of Flink.

No extra configuration is needed. Please make sure the credential file is accessible by Flink.

For example, `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`.

You need to download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it in the Flink classpath.

tengqm · 2024-12-22T04:33:23Z

...c/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactory.java

+        FactoryUtils.createCatalogFactoryHelper(this, context);
+    return new GravitinoIcebergCatalog(
+        context.getName(),
+        helper.getOptions().get(GravitinoIcebergCatalogFactoryOptions.DEFAULT_DATABASE),


Do we have to use such a long name?
Maybe CatalogFactoryOptions or even FactoryOptions is enough in such a well-defined package hierarchy?

tengqm · 2024-12-22T04:34:07Z

...c/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactory.java

+   */
+  @Override
+  public String gravitinoCatalogProvider() {
+    return "lakehouse-iceberg";


Is there a constant for this string?

There is no constant, this value comes from the catalog provider, and there is no unified definition as a constant

tengqm · 2024-12-22T04:35:43Z

...k/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConstants.java

+
+  String GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE = "hive";
+
+  @VisibleForTesting String ICEBERG_CATALOG_BACKEND_REST = CatalogUtil.ICEBERG_CATALOG_TYPE_REST;


Most, if not all fields in this interface are constants, right?

Yes, they are all constants

coolderli · 2024-12-23T01:27:21Z

...k/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConverter.java

+    all.put(
+        CommonCatalogOptions.CATALOG_TYPE.key(), GravitinoIcebergCatalogFactoryOptions.IDENTIFIER);
+    // Map "catalog-backend" to "catalog-type".
+    String catalogBackend = all.remove(IcebergConstants.CATALOG_BACKEND);


We can maintain a mapping list.

FANNG1 · 2024-12-24T09:37:16Z

docs/flink-connector/flink-catalog-iceberg.md

+
+## Storage
+
+### S3


Have you tested S3 GCS OSS storage?

FANNG1 · 2024-12-26T13:57:17Z

Hi, @sunxiaojian , Sorry for the delay, I'm working on the issues to release 0.8, may doesn't have enough time to review this PR these days.

sunxiaojian changed the title ~~[#3515]feat(flink-connector)Support flink iceberg catalog~~ [#3515] feat(flink-connector): Support flink iceberg catalog Dec 19, 2024

sunxiaojian force-pushed the support-flink-iceberg-catalog branch 3 times, most recently from e5cb05a to 24cd6b8 Compare December 19, 2024 07:34

Support flink iceberg catalog apache#3515

ef294ba

sunxiaojian force-pushed the support-flink-iceberg-catalog branch from 24cd6b8 to ef294ba Compare December 19, 2024 07:37

tengqm reviewed Dec 22, 2024

View reviewed changes

coolderli reviewed Dec 23, 2024

View reviewed changes

FANNG1 reviewed Dec 26, 2024

View reviewed changes

docs/flink-connector/flink-catalog-iceberg.md

## Storage

### S3

Copy link

Contributor

FANNG1 Dec 24, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Have you tested S3 GCS OSS storage?

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[#3515] feat(flink-connector): Support flink iceberg catalog #5914

[#3515] feat(flink-connector): Support flink iceberg catalog #5914

sunxiaojian commented Dec 19, 2024

sunxiaojian commented Dec 19, 2024

FANNG1 commented Dec 19, 2024

sunxiaojian commented Dec 19, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

sunxiaojian Dec 22, 2024

tengqm Dec 22, 2024

sunxiaojian Dec 22, 2024

coolderli Dec 23, 2024

FANNG1 Dec 24, 2024

FANNG1 commented Dec 26, 2024

		The Apache Gravitino Flink connector offers the capability to read and write Iceberg tables, with the metadata managed by the Gravitino server. To enable the use of the Iceberg catalog within the Flink connector, you must download the Iceberg Flink runtime JAR to the Flink classpath.

-The Apache Gravitino Flink connector offers the capability to read and write Iceberg tables, with the metadata managed by the Gravitino server. To enable the use of the Iceberg catalog within the Flink connector, you must download the Iceberg Flink runtime JAR to the Flink classpath.
+The Apache Gravitino Flink connector can be used to read and write Iceberg tables, with the metadata managed by the Gravitino server.
+To enable the Flink connector, you must download the Iceberg Flink runtime JAR and place it in the Flink classpath.

	#### Not supported operations:
	#### Operations Not Supported:


		## Catalog properties

		Gravitino Flink connector will transform the following property names defined in catalog properties to Flink Iceberg connector configuration.

	Gravitino Flink connector will transform the following property names defined in catalog properties to Flink Iceberg connector configuration.
	The Gravitino Flink connector transforms the following properties in a catalog to Flink connector configuration.


		### S3

		You need to add s3 secret to the Flink configuration using `s3.access-key-id` and `s3.secret-access-key`. Additionally, download the [Iceberg AWS bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-aws-bundle) and place it in the classpath of Flink.


		### OSS

		You need to add OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`. Additionally, download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip) and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` in the classpath of Flink.

-You need to add OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`. Additionally, download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip) and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` in the classpath of Flink.
+You need to add an OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`.
+Additionally, you need download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip),
+and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` to the Flink classpath.


		### GCS

		No extra configuration is needed. Please make sure the credential file is accessible by Flink, like using `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`, and download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it to the classpath of Flink.

-No extra configuration is needed. Please make sure the credential file is accessible by Flink, like using `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`, and download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it to the classpath of Flink.
+No extra configuration is needed. Please make sure the credential file is accessible by Flink.
+For example, `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`.
+You need to download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it in the Flink classpath.


		String GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE = "hive";

		@VisibleForTesting String ICEBERG_CATALOG_BACKEND_REST = CatalogUtil.ICEBERG_CATALOG_TYPE_REST;

[#3515] feat(flink-connector): Support flink iceberg catalog #5914

Are you sure you want to change the base?

[#3515] feat(flink-connector): Support flink iceberg catalog #5914

Conversation

sunxiaojian commented Dec 19, 2024

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

sunxiaojian commented Dec 19, 2024

FANNG1 commented Dec 19, 2024

sunxiaojian commented Dec 19, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

FANNG1 commented Dec 26, 2024