You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2020/08/17 12:58:21 UTC
[GitHub] [iceberg] openinx opened a new pull request #1348: Flink: Support table sink.
openinx opened a new pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348
This patch will wrap the flink's `DataStream` as a `StreamTable`, which could allow user to use SQL to insert records to iceberg table, it will try to provide the similar experience with spark sql. Currently, this patch is depending on https://github.com/apache/iceberg/pull/1185 .
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482660551
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
##########
@@ -155,16 +158,29 @@ public Builder tableSchema(TableSchema newTableSchema) {
return this;
}
+ public Builder overwrite(boolean newOverwrite) {
+ this.overwrite = newOverwrite;
+ return this;
+ }
+
@SuppressWarnings("unchecked")
public DataStreamSink<RowData> build() {
Preconditions.checkArgument(rowDataInput != null,
"Please use forRowData() to initialize the input DataStream.");
- Preconditions.checkNotNull(table, "Table shouldn't be null");
Preconditions.checkNotNull(tableLoader, "Table loader shouldn't be null");
Preconditions.checkNotNull(hadoopConf, "Hadoop configuration shouldn't be null");
+ if (table == null) {
+ tableLoader.open(hadoopConf);
+ try (TableLoader loader = tableLoader) {
+ this.table = loader.loadTable();
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to load iceberg table.", e);
Review comment:
Defining the `toString` sounds good to me.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r481625868
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java
##########
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.util.Arrays;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.sinks.AppendStreamTableSink;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.types.DataType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.sink.FlinkSink;
+
+public class IcebergTableSink implements AppendStreamTableSink<RowData> {
+ private final TableIdentifier tableIdentifier;
+ private final Table table;
+ private final CatalogLoader catalogLoader;
+ private final TableSchema tableSchema;
+ private final Configuration hadoopConf;
+
+ public IcebergTableSink(TableIdentifier tableIdentifier, Table table,
+ CatalogLoader catalogLoader, Configuration hadoopConf,
+ TableSchema tableSchema) {
+ this.tableIdentifier = tableIdentifier;
+ this.table = table;
+ this.catalogLoader = catalogLoader;
+ this.hadoopConf = hadoopConf;
+ this.tableSchema = tableSchema;
+ }
+
+ @Override
+ public DataStreamSink<?> consumeDataStream(DataStream<RowData> dataStream) {
+ return FlinkSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(TableLoader.fromCatalog(catalogLoader, tableIdentifier))
+ .hadoopConf(hadoopConf)
+ .tableSchema(tableSchema)
+ .build();
+ }
+
+ @Override
+ public DataType getConsumedDataType() {
+ return tableSchema.toRowDataType().bridgedTo(RowData.class);
+ }
+
+ @Override
+ public TableSchema getTableSchema() {
+ return this.tableSchema;
+ }
+
+ @Override
+ public TableSink<RowData> configure(String[] fieldNames, TypeInformation<?>[] fieldTypes) {
+ if (!Arrays.equals(tableSchema.getFieldNames(), fieldNames)) {
Review comment:
This is a deprecated method, no one will call it, you can just `return this`.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.util.FiniteTestSource;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.util.DataFormatConverters;
+import org.apache.flink.table.runtime.typeutils.RowDataTypeInfo;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.Pair;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSink extends AbstractTestBase {
+ private static final Configuration CONF = new Configuration();
+ private static final DataFormatConverters.RowConverter CONVERTER = new DataFormatConverters.RowConverter(
+ SimpleDataUtil.FLINK_SCHEMA.getFieldDataTypes());
+
+ private static final String TABLE_NAME = "flink_table";
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+ private String tablePath;
+ private String warehouse;
+ private Map<String, String> properties;
+ private Catalog catalog;
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+
+ private final FileFormat format;
+ private final int parallelism;
+
+ @Parameterized.Parameters(name = "{index}: format={0}, parallelism={2}")
+ public static Iterable<Object[]> data() {
+ return Arrays.asList(
+ new Object[] {"avro", 1},
+ new Object[] {"avro", 2},
+ new Object[] {"orc", 1},
+ new Object[] {"orc", 2},
+ new Object[] {"parquet", 1},
+ new Object[] {"parquet", 2}
+ );
+ }
+
+ public TestFlinkTableSink(String format, int parallelism) {
+ this.format = FileFormat.valueOf(format.toUpperCase(Locale.ENGLISH));
+ this.parallelism = parallelism;
+ }
+
+ @Before
+ public void before() throws IOException {
+ File folder = tempFolder.newFolder();
+ warehouse = folder.getAbsolutePath();
+
+ tablePath = warehouse.concat("/default/").concat(TABLE_NAME);
+ Assert.assertTrue("Should create the table path correctly.", new File(tablePath).mkdirs());
+
+ properties = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name());
+ catalog = new HadoopCatalog(CONF, warehouse);
+
+ env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ env.enableCheckpointing(400);
+ env.setParallelism(parallelism);
+
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inStreamingMode()
+ .build();
+ tEnv = StreamTableEnvironment.create(env, settings);
+ tEnv.executeSql(String.format("create catalog iceberg_catalog with (" +
+ "'type'='iceberg', 'catalog-type'='hadoop', 'warehouse'='%s')", warehouse));
+ tEnv.executeSql("use catalog iceberg_catalog");
+ tEnv.getConfig().getConfiguration().set(TableConfigOptions.TABLE_DYNAMIC_TABLE_OPTIONS_ENABLED, true);
Review comment:
Looks like there is no dynamic table options. (Table hints)
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.util.FiniteTestSource;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.util.DataFormatConverters;
+import org.apache.flink.table.runtime.typeutils.RowDataTypeInfo;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.Pair;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSink extends AbstractTestBase {
+ private static final Configuration CONF = new Configuration();
+ private static final DataFormatConverters.RowConverter CONVERTER = new DataFormatConverters.RowConverter(
+ SimpleDataUtil.FLINK_SCHEMA.getFieldDataTypes());
+
+ private static final String TABLE_NAME = "flink_table";
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+ private String tablePath;
+ private String warehouse;
+ private Map<String, String> properties;
+ private Catalog catalog;
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+
+ private final FileFormat format;
+ private final int parallelism;
+
+ @Parameterized.Parameters(name = "{index}: format={0}, parallelism={2}")
+ public static Iterable<Object[]> data() {
+ return Arrays.asList(
+ new Object[] {"avro", 1},
+ new Object[] {"avro", 2},
+ new Object[] {"orc", 1},
+ new Object[] {"orc", 2},
+ new Object[] {"parquet", 1},
+ new Object[] {"parquet", 2}
+ );
+ }
+
+ public TestFlinkTableSink(String format, int parallelism) {
+ this.format = FileFormat.valueOf(format.toUpperCase(Locale.ENGLISH));
+ this.parallelism = parallelism;
+ }
+
+ @Before
+ public void before() throws IOException {
+ File folder = tempFolder.newFolder();
+ warehouse = folder.getAbsolutePath();
+
+ tablePath = warehouse.concat("/default/").concat(TABLE_NAME);
+ Assert.assertTrue("Should create the table path correctly.", new File(tablePath).mkdirs());
+
+ properties = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name());
+ catalog = new HadoopCatalog(CONF, warehouse);
+
+ env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ env.enableCheckpointing(400);
+ env.setParallelism(parallelism);
+
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inStreamingMode()
Review comment:
Can we use Parameterized for batch too?
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.util.FiniteTestSource;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.util.DataFormatConverters;
+import org.apache.flink.table.runtime.typeutils.RowDataTypeInfo;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.Pair;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSink extends AbstractTestBase {
+ private static final Configuration CONF = new Configuration();
+ private static final DataFormatConverters.RowConverter CONVERTER = new DataFormatConverters.RowConverter(
+ SimpleDataUtil.FLINK_SCHEMA.getFieldDataTypes());
+
+ private static final String TABLE_NAME = "flink_table";
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+ private String tablePath;
+ private String warehouse;
+ private Map<String, String> properties;
+ private Catalog catalog;
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+
+ private final FileFormat format;
+ private final int parallelism;
+
+ @Parameterized.Parameters(name = "{index}: format={0}, parallelism={2}")
+ public static Iterable<Object[]> data() {
+ return Arrays.asList(
+ new Object[] {"avro", 1},
+ new Object[] {"avro", 2},
+ new Object[] {"orc", 1},
+ new Object[] {"orc", 2},
+ new Object[] {"parquet", 1},
+ new Object[] {"parquet", 2}
+ );
+ }
+
+ public TestFlinkTableSink(String format, int parallelism) {
+ this.format = FileFormat.valueOf(format.toUpperCase(Locale.ENGLISH));
+ this.parallelism = parallelism;
+ }
+
+ @Before
+ public void before() throws IOException {
+ File folder = tempFolder.newFolder();
+ warehouse = folder.getAbsolutePath();
+
+ tablePath = warehouse.concat("/default/").concat(TABLE_NAME);
+ Assert.assertTrue("Should create the table path correctly.", new File(tablePath).mkdirs());
+
+ properties = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name());
+ catalog = new HadoopCatalog(CONF, warehouse);
+
+ env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ env.enableCheckpointing(400);
+ env.setParallelism(parallelism);
+
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inStreamingMode()
+ .build();
+ tEnv = StreamTableEnvironment.create(env, settings);
+ tEnv.executeSql(String.format("create catalog iceberg_catalog with (" +
+ "'type'='iceberg', 'catalog-type'='hadoop', 'warehouse'='%s')", warehouse));
+ tEnv.executeSql("use catalog iceberg_catalog");
+ tEnv.getConfig().getConfiguration().set(TableConfigOptions.TABLE_DYNAMIC_TABLE_OPTIONS_ENABLED, true);
+
+ catalog.createTable(TableIdentifier.parse("default." + TABLE_NAME),
+ SimpleDataUtil.SCHEMA,
+ PartitionSpec.unpartitioned(),
+ properties);
+ }
+
+ private DataStream<RowData> generateInputStream(List<Row> rows) {
+ TypeInformation<Row> typeInformation = new RowTypeInfo(SimpleDataUtil.FLINK_SCHEMA.getFieldTypes());
+ return env.addSource(new FiniteTestSource<>(rows), typeInformation)
+ .map(CONVERTER::toInternal, RowDataTypeInfo.of(SimpleDataUtil.ROW_TYPE));
+ }
+
+ private Pair<List<Row>, List<Record>> generateData() {
+ String[] worlds = new String[] {"hello", "world", "foo", "bar", "apache", "foundation"};
+ List<Row> rows = Lists.newArrayList();
+ List<Record> expected = Lists.newArrayList();
+ for (int i = 0; i < worlds.length; i++) {
+ rows.add(Row.of(i + 1, worlds[i]));
+ Record record = SimpleDataUtil.createRecord(i + 1, worlds[i]);
+ expected.add(record);
+ expected.add(record);
+ }
+ return Pair.of(rows, expected);
+ }
+
+ @Test
+ public void testStreamSQL() throws Exception {
+ Pair<List<Row>, List<Record>> data = generateData();
+ List<Row> rows = data.first();
+ List<Record> expected = data.second();
+ DataStream<RowData> stream = generateInputStream(rows);
+
+ // Register the rows into a temporary table named 'sourceTable'.
+ tEnv.createTemporaryView("sourceTable", tEnv.fromDataStream(stream, $("id"), $("data")));
Review comment:
Can we use `TableEnvironment.fromValues`?
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java
##########
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.util.Arrays;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.sinks.AppendStreamTableSink;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.types.DataType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.sink.FlinkSink;
+
+public class IcebergTableSink implements AppendStreamTableSink<RowData> {
Review comment:
We can add TODO for these interfaces:
Implement `OverwritableTableSink`, so in the Flink SQL, user can write these SQLs:
`INSERT OVERWRITE t ...`
Implement `PartitionableTableSink`, user can write:
`INSERT OVERWRITE/INTO t PARTITION(...)`
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.util.FiniteTestSource;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.util.DataFormatConverters;
+import org.apache.flink.table.runtime.typeutils.RowDataTypeInfo;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.Pair;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSink extends AbstractTestBase {
+ private static final Configuration CONF = new Configuration();
+ private static final DataFormatConverters.RowConverter CONVERTER = new DataFormatConverters.RowConverter(
+ SimpleDataUtil.FLINK_SCHEMA.getFieldDataTypes());
+
+ private static final String TABLE_NAME = "flink_table";
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+ private String tablePath;
+ private String warehouse;
+ private Map<String, String> properties;
+ private Catalog catalog;
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+
+ private final FileFormat format;
+ private final int parallelism;
+
+ @Parameterized.Parameters(name = "{index}: format={0}, parallelism={2}")
+ public static Iterable<Object[]> data() {
+ return Arrays.asList(
+ new Object[] {"avro", 1},
+ new Object[] {"avro", 2},
+ new Object[] {"orc", 1},
+ new Object[] {"orc", 2},
+ new Object[] {"parquet", 1},
+ new Object[] {"parquet", 2}
+ );
+ }
+
+ public TestFlinkTableSink(String format, int parallelism) {
+ this.format = FileFormat.valueOf(format.toUpperCase(Locale.ENGLISH));
+ this.parallelism = parallelism;
+ }
+
+ @Before
+ public void before() throws IOException {
+ File folder = tempFolder.newFolder();
+ warehouse = folder.getAbsolutePath();
+
+ tablePath = warehouse.concat("/default/").concat(TABLE_NAME);
+ Assert.assertTrue("Should create the table path correctly.", new File(tablePath).mkdirs());
+
+ properties = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name());
+ catalog = new HadoopCatalog(CONF, warehouse);
+
+ env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ env.enableCheckpointing(400);
+ env.setParallelism(parallelism);
+
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inStreamingMode()
+ .build();
+ tEnv = StreamTableEnvironment.create(env, settings);
+ tEnv.executeSql(String.format("create catalog iceberg_catalog with (" +
+ "'type'='iceberg', 'catalog-type'='hadoop', 'warehouse'='%s')", warehouse));
+ tEnv.executeSql("use catalog iceberg_catalog");
+ tEnv.getConfig().getConfiguration().set(TableConfigOptions.TABLE_DYNAMIC_TABLE_OPTIONS_ENABLED, true);
+
+ catalog.createTable(TableIdentifier.parse("default." + TABLE_NAME),
+ SimpleDataUtil.SCHEMA,
+ PartitionSpec.unpartitioned(),
+ properties);
+ }
+
+ private DataStream<RowData> generateInputStream(List<Row> rows) {
+ TypeInformation<Row> typeInformation = new RowTypeInfo(SimpleDataUtil.FLINK_SCHEMA.getFieldTypes());
+ return env.addSource(new FiniteTestSource<>(rows), typeInformation)
+ .map(CONVERTER::toInternal, RowDataTypeInfo.of(SimpleDataUtil.ROW_TYPE));
+ }
+
+ private Pair<List<Row>, List<Record>> generateData() {
+ String[] worlds = new String[] {"hello", "world", "foo", "bar", "apache", "foundation"};
+ List<Row> rows = Lists.newArrayList();
+ List<Record> expected = Lists.newArrayList();
+ for (int i = 0; i < worlds.length; i++) {
+ rows.add(Row.of(i + 1, worlds[i]));
+ Record record = SimpleDataUtil.createRecord(i + 1, worlds[i]);
+ expected.add(record);
+ expected.add(record);
+ }
+ return Pair.of(rows, expected);
+ }
+
+ @Test
+ public void testStreamSQL() throws Exception {
+ Pair<List<Row>, List<Record>> data = generateData();
+ List<Row> rows = data.first();
+ List<Record> expected = data.second();
+ DataStream<RowData> stream = generateInputStream(rows);
+
+ // Register the rows into a temporary table named 'sourceTable'.
+ tEnv.createTemporaryView("sourceTable", tEnv.fromDataStream(stream, $("id"), $("data")));
+
+ // Redirect the records from source table to destination table.
+ String insertSQL = String.format("INSERT INTO %s SELECT id,data from sourceTable", TABLE_NAME);
+ TableResult result = tEnv.executeSql(insertSQL);
+ waitComplete(result);
+
+ // Assert the table records as expected.
+ SimpleDataUtil.assertTableRecords(tablePath, expected);
+ }
+
+ @Test
+ public void testBatchSQL() throws Exception {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .inBatchMode()
+ .useBlinkPlanner()
+ .build();
+ TableEnvironment batchEnv = TableEnvironment.create(settings);
+ batchEnv.executeSql(String.format("create catalog batch_catalog with (" +
+ "'type'='iceberg', 'catalog-type'='hadoop', 'warehouse'='%s')", warehouse));
+ batchEnv.executeSql("use catalog batch_catalog");
+ batchEnv.getConfig().getConfiguration().set(TableConfigOptions.TABLE_DYNAMIC_TABLE_OPTIONS_ENABLED, true);
+
+ // Create source table.
+ catalog.createTable(TableIdentifier.parse("default.sourceTable"),
+ SimpleDataUtil.SCHEMA,
+ PartitionSpec.unpartitioned(),
+ properties);
+
+ TableResult result;
+ String[] words = new String[] {"hello", "world", "apache"};
+ List<Record> expected = Lists.newArrayList();
+ for (int i = 0; i < words.length; i++) {
+ expected.add(SimpleDataUtil.createRecord(i, words[i]));
+ result = batchEnv.executeSql(String.format("INSERT INTO sourceTable SELECT %d, '%s'", i, words[i]));
+ waitComplete(result);
+ }
+
+ // Assert the table records as expected.
+ SimpleDataUtil.assertTableRecords(warehouse.concat("/default/sourceTable"), expected);
+ }
+
+ private static void waitComplete(TableResult result) {
Review comment:
You can just add a method like:
```
def execInsertSqlAndWaitResult(tEnv: TableEnvironment, insert: String): JobExecutionResult = {
tEnv.executeSql(insert).getJobClient.get
.getJobExecutionResult(Thread.currentThread.getContextClassLoader)
.get
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482688895
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
##########
@@ -164,16 +168,51 @@ private void commitUpToCheckpoint(long checkpointId) {
pendingDataFiles.addAll(dataFiles);
}
- AppendFiles appendFiles = table.newAppend();
- pendingDataFiles.forEach(appendFiles::appendFile);
- appendFiles.set(MAX_COMMITTED_CHECKPOINT_ID, Long.toString(checkpointId));
- appendFiles.set(FLINK_JOB_ID, flinkJobId);
- appendFiles.commit();
+ if (replacePartitions) {
+ replacePartitions(pendingDataFiles, checkpointId);
+ } else {
+ append(pendingDataFiles, checkpointId);
+ }
// Clear the committed data files from dataFilesPerCheckpoint.
pendingFileMap.clear();
}
+ private void replacePartitions(List<DataFile> dataFiles, long checkpointId) {
+ ReplacePartitions dynamicOverwrite = table.newReplacePartitions();
Review comment:
Yes, Flink's semantics are to replace partitions for overwrite, here is the [1](https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/table/sinks/OverwritableTableSink.html), [2](https://ci.apache.org/projects/flink/flink-docs-release-1.9/api/java/org/apache/flink/table/sinks/PartitionableTableSink.html)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482693653
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -126,12 +126,16 @@ public static void assertTableRows(String tablePath, List<RowData> expected) thr
assertTableRecords(tablePath, expectedRecords);
}
- public static void assertTableRecords(String tablePath, List<Record> expected) throws IOException {
- Preconditions.checkArgument(expected != null, "expected records shouldn't be null");
- Table newTable = new HadoopTables().load(tablePath);
- try (CloseableIterable<Record> iterable = IcebergGenerics.read(newTable).build()) {
+ public static void assertTableRecords(Table table, List<Record> expected) throws IOException {
+ table.refresh();
Review comment:
Since we don't support to scan table by flink sql, so we have to read records from iceberg table by iceberg Java API in unit tests. In this [test](https://github.com/apache/iceberg/pull/1348/files#diff-6fc1fa50af5a93944270e7dd4c0ea393R98), we get the `icebergTable` instance firstly, then the following test methods will commit the iceberg table by flink sql, the `icebergTable` need a fresh to catch the latest changes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r481786875
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkTableFactory.java
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.StreamTableSinkFactory;
+import org.apache.flink.table.sinks.StreamTableSink;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+
+public class FlinkTableFactory implements StreamTableSinkFactory<RowData> {
+ private final FlinkCatalog catalog;
+
+ public FlinkTableFactory(FlinkCatalog catalog) {
+ this.catalog = catalog;
+ }
+
+ @Override
+ public StreamTableSink<RowData> createTableSink(Context context) {
+ ObjectIdentifier identifier = context.getObjectIdentifier();
+ ObjectPath objectPath = new ObjectPath(identifier.getDatabaseName(), identifier.getObjectName());
+ TableIdentifier icebergIdentifier = catalog.toIdentifier(objectPath);
+ try {
+ Table table = catalog.loadIcebergTable(objectPath);
+ return new IcebergTableSink(icebergIdentifier, table,
Review comment:
Make sense to me, we also don't need to pass the `icebergIdentifier` to `IcebergTableSink`, that makes code more simplier.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482589889
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
##########
@@ -155,16 +158,29 @@ public Builder tableSchema(TableSchema newTableSchema) {
return this;
}
+ public Builder overwrite(boolean newOverwrite) {
+ this.overwrite = newOverwrite;
+ return this;
+ }
+
@SuppressWarnings("unchecked")
public DataStreamSink<RowData> build() {
Preconditions.checkArgument(rowDataInput != null,
"Please use forRowData() to initialize the input DataStream.");
- Preconditions.checkNotNull(table, "Table shouldn't be null");
Preconditions.checkNotNull(tableLoader, "Table loader shouldn't be null");
Preconditions.checkNotNull(hadoopConf, "Hadoop configuration shouldn't be null");
+ if (table == null) {
+ tableLoader.open(hadoopConf);
+ try (TableLoader loader = tableLoader) {
+ this.table = loader.loadTable();
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to load iceberg table.", e);
Review comment:
Minor: it would be nice to have more context here. Maybe the table loader should define a `toString` that could be used in the error message here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482082688
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.Expressions;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.After;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSink extends FlinkCatalogTestBase {
+ private static final String TABLE_NAME = "test_table";
+ private TableEnvironment tEnv;
+ private org.apache.iceberg.Table icebergTable;
+
+ private final FileFormat format;
+ private final boolean isStreamingJob;
+
+ @Parameterized.Parameters(name = "{index}: format={0}, isStreaming={1}, catalogName={2}, baseNamespace={3}")
+ public static Iterable<Object[]> parameters() {
+ List<Object[]> parameters = Lists.newArrayList();
+ for (FileFormat format : new FileFormat[] {FileFormat.ORC, FileFormat.AVRO, FileFormat.PARQUET}) {
+ for (Boolean isStreaming : new Boolean[] {true, false}) {
+ for (Object[] catalogParams : FlinkCatalogTestBase.parameters()) {
+ String catalogName = (String) catalogParams[0];
+ String[] baseNamespace = (String[]) catalogParams[1];
+ parameters.add(new Object[] {format, isStreaming, catalogName, baseNamespace});
+ }
+ }
+ }
+ return parameters;
+ }
+
+ public TestFlinkTableSink(FileFormat format, Boolean isStreamingJob, String catalogName, String[] baseNamespace) {
+ super(catalogName, baseNamespace);
+ this.format = format;
+ this.isStreamingJob = isStreamingJob;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreamingJob) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ env.enableCheckpointing(400);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @Before
+ public void before() {
+ sql("CREATE DATABASE %s", flinkDatabase);
+ sql("USE CATALOG %s", catalogName);
+ sql("USE %s", DATABASE);
+
+ Map<String, String> properties = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name());
+ this.icebergTable = validationCatalog
Review comment:
We could use flink DDL to create table here if https://github.com/apache/iceberg/pull/1393 get merged.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#issuecomment-683352413
@openinx, can you rebase this? I think it is next to review, right?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482592010
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
##########
@@ -164,16 +168,51 @@ private void commitUpToCheckpoint(long checkpointId) {
pendingDataFiles.addAll(dataFiles);
}
- AppendFiles appendFiles = table.newAppend();
- pendingDataFiles.forEach(appendFiles::appendFile);
- appendFiles.set(MAX_COMMITTED_CHECKPOINT_ID, Long.toString(checkpointId));
- appendFiles.set(FLINK_JOB_ID, flinkJobId);
- appendFiles.commit();
+ if (replacePartitions) {
+ replacePartitions(pendingDataFiles, checkpointId);
+ } else {
+ append(pendingDataFiles, checkpointId);
+ }
// Clear the committed data files from dataFilesPerCheckpoint.
pendingFileMap.clear();
}
+ private void replacePartitions(List<DataFile> dataFiles, long checkpointId) {
+ ReplacePartitions dynamicOverwrite = table.newReplacePartitions();
Review comment:
I just want to note that we don't encourage the use of `ReplacePartitions` because the data it deletes is implicit. It is better to specify what data should be overwritten, like in the new API for Spark:
```scala
df.writeTo("iceberg.db.table").overwrite($"date" === "2020-09-01")
```
If Flink's semantics are to replace partitions for overwrite, then it should be okay. But I highly recommend being more explicit about data replacement.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r481999310
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.Expressions;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSink extends AbstractTestBase {
Review comment:
Think about this unit test again, we'd better to extend the `FlinkCatalogTestBase` so that we could cover both hive and hadoop catalog cases.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#issuecomment-684814404
Ping @rdblue @JingsongLi for reviewing, Thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r481803458
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java
##########
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.util.Arrays;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.sinks.AppendStreamTableSink;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.types.DataType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.sink.FlinkSink;
+
+public class IcebergTableSink implements AppendStreamTableSink<RowData> {
+ private final TableIdentifier tableIdentifier;
+ private final Table table;
+ private final CatalogLoader catalogLoader;
+ private final TableSchema tableSchema;
+ private final Configuration hadoopConf;
+
+ public IcebergTableSink(TableIdentifier tableIdentifier, Table table,
+ CatalogLoader catalogLoader, Configuration hadoopConf,
+ TableSchema tableSchema) {
+ this.tableIdentifier = tableIdentifier;
+ this.table = table;
+ this.catalogLoader = catalogLoader;
+ this.hadoopConf = hadoopConf;
+ this.tableSchema = tableSchema;
+ }
+
+ @Override
+ public DataStreamSink<?> consumeDataStream(DataStream<RowData> dataStream) {
+ return FlinkSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(TableLoader.fromCatalog(catalogLoader, tableIdentifier))
+ .hadoopConf(hadoopConf)
+ .tableSchema(tableSchema)
+ .build();
+ }
+
+ @Override
+ public DataType getConsumedDataType() {
+ return tableSchema.toRowDataType().bridgedTo(RowData.class);
+ }
+
+ @Override
+ public TableSchema getTableSchema() {
+ return this.tableSchema;
+ }
+
+ @Override
+ public TableSink<RowData> configure(String[] fieldNames, TypeInformation<?>[] fieldTypes) {
+ if (!Arrays.equals(tableSchema.getFieldNames(), fieldNames)) {
Review comment:
OK, I see. will do .
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482688895
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
##########
@@ -164,16 +168,51 @@ private void commitUpToCheckpoint(long checkpointId) {
pendingDataFiles.addAll(dataFiles);
}
- AppendFiles appendFiles = table.newAppend();
- pendingDataFiles.forEach(appendFiles::appendFile);
- appendFiles.set(MAX_COMMITTED_CHECKPOINT_ID, Long.toString(checkpointId));
- appendFiles.set(FLINK_JOB_ID, flinkJobId);
- appendFiles.commit();
+ if (replacePartitions) {
+ replacePartitions(pendingDataFiles, checkpointId);
+ } else {
+ append(pendingDataFiles, checkpointId);
+ }
// Clear the committed data files from dataFilesPerCheckpoint.
pendingFileMap.clear();
}
+ private void replacePartitions(List<DataFile> dataFiles, long checkpointId) {
+ ReplacePartitions dynamicOverwrite = table.newReplacePartitions();
Review comment:
Yes, Flink's semantics are to replace partitions for overwrite, here is the [1](https://javadoc.io/doc/org.apache.flink/flink-table-common/latest/index.html), [2](https://ci.apache.org/projects/flink/flink-docs-release-1.9/api/java/org/apache/flink/table/sinks/PartitionableTableSink.html)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482688895
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
##########
@@ -164,16 +168,51 @@ private void commitUpToCheckpoint(long checkpointId) {
pendingDataFiles.addAll(dataFiles);
}
- AppendFiles appendFiles = table.newAppend();
- pendingDataFiles.forEach(appendFiles::appendFile);
- appendFiles.set(MAX_COMMITTED_CHECKPOINT_ID, Long.toString(checkpointId));
- appendFiles.set(FLINK_JOB_ID, flinkJobId);
- appendFiles.commit();
+ if (replacePartitions) {
+ replacePartitions(pendingDataFiles, checkpointId);
+ } else {
+ append(pendingDataFiles, checkpointId);
+ }
// Clear the committed data files from dataFilesPerCheckpoint.
pendingFileMap.clear();
}
+ private void replacePartitions(List<DataFile> dataFiles, long checkpointId) {
+ ReplacePartitions dynamicOverwrite = table.newReplacePartitions();
Review comment:
Yes, Flink's semantics are to replace partitions for overwrite, here is the [document](https://ci.apache.org/projects/flink/flink-docs-release-1.9/api/java/org/apache/flink/table/sinks/PartitionableTableSink.html)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue merged pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
rdblue merged pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r479859466
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.sink;
+
+import java.util.Locale;
+import java.util.Map;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.util.DataFormatConverters;
+import org.apache.flink.table.runtime.typeutils.RowDataTypeInfo;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.utils.TypeConversions;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.PropertyUtil;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES;
+import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT;
+
+public class FlinkSink {
+
+ private static final String ICEBERG_STREAM_WRITER_NAME = IcebergStreamWriter.class.getSimpleName();
+ private static final String ICEBERG_FILES_COMMITTER_NAME = IcebergFilesCommitter.class.getSimpleName();
+
+ private FlinkSink() {
+ }
+
+ public static Builder forRow(DataStream<Row> input) {
+ return new Builder().forRow(input);
+ }
+
+ public static Builder forRowData(DataStream<RowData> input) {
+ return new Builder().forRowData(input);
+ }
+
+ public static class Builder {
+ private DataStream<Row> rowInput = null;
+ private DataStream<RowData> rowDataInput = null;
+ private TableLoader tableLoader;
+ private Configuration hadoopConf;
+ private Table table;
+ private TableSchema tableSchema;
+
+ private Builder forRow(DataStream<Row> newRowInput) {
+ this.rowInput = newRowInput;
+ return this;
+ }
+
+ private Builder forRowData(DataStream<RowData> newRowDataInput) {
+ this.rowDataInput = newRowDataInput;
+ return this;
+ }
+
+ public Builder table(Table newTable) {
+ this.table = newTable;
+ return this;
+ }
+
+ public Builder tableLoader(TableLoader newTableLoader) {
+ this.tableLoader = newTableLoader;
+ return this;
+ }
+
+ public Builder hadoopConf(Configuration newHadoopConf) {
+ this.hadoopConf = newHadoopConf;
+ return this;
+ }
+
+ public Builder tableSchema(TableSchema newTableSchema) {
+ this.tableSchema = newTableSchema;
+ return this;
+ }
+
+ private DataStream<RowData> convert() {
+ Preconditions.checkArgument(rowInput != null, "The DataStream<Row> to convert shouldn't be null");
+
+ RowType rowType;
+ DataType[] fieldDataTypes;
+ if (tableSchema != null) {
+ rowType = (RowType) tableSchema.toRowDataType().getLogicalType();
+ fieldDataTypes = tableSchema.getFieldDataTypes();
+ } else {
+ rowType = FlinkSchemaUtil.convert(table.schema());
+ fieldDataTypes = TypeConversions.fromLogicalToDataType(rowType.getChildren().toArray(new LogicalType[0]));
+ }
+
+ DataFormatConverters.RowConverter rowConverter = new DataFormatConverters.RowConverter(fieldDataTypes);
+
+ return rowInput.map(rowConverter::toInternal, RowDataTypeInfo.of(rowType));
+ }
+
+ @SuppressWarnings("unchecked")
+ public DataStreamSink<RowData> build() {
+ Preconditions.checkArgument(rowInput != null || rowDataInput != null,
+ "Should initialize input DataStream first with DataStream<Row> or DataStream<RowData>");
+ Preconditions.checkArgument(rowInput == null || rowDataInput == null,
+ "Could only initialize input DataStream with either DataStream<Row> or DataStream<RowData>");
+ Preconditions.checkNotNull(table, "Table shouldn't be null");
+ Preconditions.checkNotNull(tableLoader, "Table loader shouldn't be null");
+ Preconditions.checkNotNull(hadoopConf, "Hadoop configuration shouldn't be null");
+
+ DataStream<RowData> inputStream = rowInput != null ? convert() : rowDataInput;
+
+ IcebergStreamWriter<RowData> streamWriter = createStreamWriter(table, tableSchema);
+ IcebergFilesCommitter filesCommitter = new IcebergFilesCommitter(tableLoader, hadoopConf);
+
+ DataStream<Void> returnStream = inputStream
+ .transform(ICEBERG_STREAM_WRITER_NAME, TypeInformation.of(DataFile.class), streamWriter)
+ .setParallelism(inputStream.getParallelism())
+ .transform(ICEBERG_FILES_COMMITTER_NAME, Types.VOID, filesCommitter)
+ .setParallelism(1)
+ .setMaxParallelism(1);
+
+ return returnStream.addSink(new DiscardingSink())
+ .name(String.format("IcebergSink %s", table.toString()))
+ .setParallelism(1);
Review comment:
You mean you want to feed the committed data files to kafka ? Is that meaningful for users ? It will be better to understand if we have such user cases I guess.
### Some context
in the first sink version, I made the `IcebergFilesCommitter` implemented the `SinkFucntion`, then we could chain the function by `addSink` directly, while we found that it did not work for bounded stream because there was no interface/method to indicate that this stream is a bounded one, then we have no way to commit those data files into iceberg table when the stream has reached its end. So we have to turn to `AbstractStreamOperator` and implemented a `BoundedOneInput` interface. Finally, int this version, we will transform the data stream twice (the first one: rowdata -> dataFiles, the second one: datafiles -> void), and finally add a discarding sink.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r481624265
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkTableFactory.java
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.StreamTableSinkFactory;
+import org.apache.flink.table.sinks.StreamTableSink;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+
+public class FlinkTableFactory implements StreamTableSinkFactory<RowData> {
+ private final FlinkCatalog catalog;
+
+ public FlinkTableFactory(FlinkCatalog catalog) {
+ this.catalog = catalog;
+ }
+
+ @Override
+ public StreamTableSink<RowData> createTableSink(Context context) {
+ ObjectIdentifier identifier = context.getObjectIdentifier();
+ ObjectPath objectPath = new ObjectPath(identifier.getDatabaseName(), identifier.getObjectName());
+ TableIdentifier icebergIdentifier = catalog.toIdentifier(objectPath);
+ try {
+ Table table = catalog.loadIcebergTable(objectPath);
+ return new IcebergTableSink(icebergIdentifier, table,
Review comment:
I think it is better to just pass a table loader to sink, source and sink can reuse this loader creation function, just like in:
https://github.com/apache/iceberg/pull/1293/files#diff-0ad7dfff9cfa32fbb760796d976fd650R61
What do you think?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#issuecomment-686089635
+1 overall. I'd merge this now even with a couple of minor comments but it appears that merging #1393 caused conflicts.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482688895
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
##########
@@ -164,16 +168,51 @@ private void commitUpToCheckpoint(long checkpointId) {
pendingDataFiles.addAll(dataFiles);
}
- AppendFiles appendFiles = table.newAppend();
- pendingDataFiles.forEach(appendFiles::appendFile);
- appendFiles.set(MAX_COMMITTED_CHECKPOINT_ID, Long.toString(checkpointId));
- appendFiles.set(FLINK_JOB_ID, flinkJobId);
- appendFiles.commit();
+ if (replacePartitions) {
+ replacePartitions(pendingDataFiles, checkpointId);
+ } else {
+ append(pendingDataFiles, checkpointId);
+ }
// Clear the committed data files from dataFilesPerCheckpoint.
pendingFileMap.clear();
}
+ private void replacePartitions(List<DataFile> dataFiles, long checkpointId) {
+ ReplacePartitions dynamicOverwrite = table.newReplacePartitions();
Review comment:
Yes, Flink's semantics are to replace partitions for overwrite, here is the [1](https://github.com/apache/iceberg/pull/1348/files#r482688895), [2](https://ci.apache.org/projects/flink/flink-docs-release-1.9/api/java/org/apache/flink/table/sinks/PartitionableTableSink.html)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r481868133
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.util.FiniteTestSource;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.util.DataFormatConverters;
+import org.apache.flink.table.runtime.typeutils.RowDataTypeInfo;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.Pair;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSink extends AbstractTestBase {
+ private static final Configuration CONF = new Configuration();
+ private static final DataFormatConverters.RowConverter CONVERTER = new DataFormatConverters.RowConverter(
+ SimpleDataUtil.FLINK_SCHEMA.getFieldDataTypes());
+
+ private static final String TABLE_NAME = "flink_table";
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+ private String tablePath;
+ private String warehouse;
+ private Map<String, String> properties;
+ private Catalog catalog;
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+
+ private final FileFormat format;
+ private final int parallelism;
+
+ @Parameterized.Parameters(name = "{index}: format={0}, parallelism={2}")
+ public static Iterable<Object[]> data() {
+ return Arrays.asList(
+ new Object[] {"avro", 1},
+ new Object[] {"avro", 2},
+ new Object[] {"orc", 1},
+ new Object[] {"orc", 2},
+ new Object[] {"parquet", 1},
+ new Object[] {"parquet", 2}
+ );
+ }
+
+ public TestFlinkTableSink(String format, int parallelism) {
+ this.format = FileFormat.valueOf(format.toUpperCase(Locale.ENGLISH));
+ this.parallelism = parallelism;
+ }
+
+ @Before
+ public void before() throws IOException {
+ File folder = tempFolder.newFolder();
+ warehouse = folder.getAbsolutePath();
+
+ tablePath = warehouse.concat("/default/").concat(TABLE_NAME);
+ Assert.assertTrue("Should create the table path correctly.", new File(tablePath).mkdirs());
+
+ properties = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name());
+ catalog = new HadoopCatalog(CONF, warehouse);
+
+ env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ env.enableCheckpointing(400);
+ env.setParallelism(parallelism);
+
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inStreamingMode()
+ .build();
+ tEnv = StreamTableEnvironment.create(env, settings);
+ tEnv.executeSql(String.format("create catalog iceberg_catalog with (" +
+ "'type'='iceberg', 'catalog-type'='hadoop', 'warehouse'='%s')", warehouse));
+ tEnv.executeSql("use catalog iceberg_catalog");
+ tEnv.getConfig().getConfiguration().set(TableConfigOptions.TABLE_DYNAMIC_TABLE_OPTIONS_ENABLED, true);
Review comment:
OK, it could be removed now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx closed pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx closed pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482594339
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.Expressions;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.After;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSink extends FlinkCatalogTestBase {
+ private static final String TABLE_NAME = "test_table";
+ private TableEnvironment tEnv;
+ private org.apache.iceberg.Table icebergTable;
+
+ private final FileFormat format;
+ private final boolean isStreamingJob;
+
+ @Parameterized.Parameters(name = "{index}: format={0}, isStreaming={1}, catalogName={2}, baseNamespace={3}")
+ public static Iterable<Object[]> parameters() {
+ List<Object[]> parameters = Lists.newArrayList();
+ for (FileFormat format : new FileFormat[] {FileFormat.ORC, FileFormat.AVRO, FileFormat.PARQUET}) {
+ for (Boolean isStreaming : new Boolean[] {true, false}) {
+ for (Object[] catalogParams : FlinkCatalogTestBase.parameters()) {
+ String catalogName = (String) catalogParams[0];
+ String[] baseNamespace = (String[]) catalogParams[1];
+ parameters.add(new Object[] {format, isStreaming, catalogName, baseNamespace});
+ }
+ }
+ }
+ return parameters;
+ }
+
+ public TestFlinkTableSink(FileFormat format, Boolean isStreamingJob, String catalogName, String[] baseNamespace) {
+ super(catalogName, baseNamespace);
+ this.format = format;
+ this.isStreamingJob = isStreamingJob;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreamingJob) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ env.enableCheckpointing(400);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @Before
+ public void before() {
+ sql("CREATE DATABASE %s", flinkDatabase);
+ sql("USE CATALOG %s", catalogName);
+ sql("USE %s", DATABASE);
+
+ Map<String, String> properties = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name());
+ this.icebergTable = validationCatalog
Review comment:
It was merged!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#issuecomment-683510188
@rdblue Yeah, Thanks for the merging, let me rebase this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r481800975
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java
##########
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.util.Arrays;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.sinks.AppendStreamTableSink;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.types.DataType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.sink.FlinkSink;
+
+public class IcebergTableSink implements AppendStreamTableSink<RowData> {
Review comment:
Thanks for the remainding, before we add the TODO comment, I will try to implement those two interfaces in the next path.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r481815513
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.util.FiniteTestSource;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.util.DataFormatConverters;
+import org.apache.flink.table.runtime.typeutils.RowDataTypeInfo;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.Pair;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSink extends AbstractTestBase {
+ private static final Configuration CONF = new Configuration();
+ private static final DataFormatConverters.RowConverter CONVERTER = new DataFormatConverters.RowConverter(
+ SimpleDataUtil.FLINK_SCHEMA.getFieldDataTypes());
+
+ private static final String TABLE_NAME = "flink_table";
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+ private String tablePath;
+ private String warehouse;
+ private Map<String, String> properties;
+ private Catalog catalog;
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+
+ private final FileFormat format;
+ private final int parallelism;
+
+ @Parameterized.Parameters(name = "{index}: format={0}, parallelism={2}")
+ public static Iterable<Object[]> data() {
+ return Arrays.asList(
+ new Object[] {"avro", 1},
+ new Object[] {"avro", 2},
+ new Object[] {"orc", 1},
+ new Object[] {"orc", 2},
+ new Object[] {"parquet", 1},
+ new Object[] {"parquet", 2}
+ );
+ }
+
+ public TestFlinkTableSink(String format, int parallelism) {
+ this.format = FileFormat.valueOf(format.toUpperCase(Locale.ENGLISH));
+ this.parallelism = parallelism;
+ }
+
+ @Before
+ public void before() throws IOException {
+ File folder = tempFolder.newFolder();
+ warehouse = folder.getAbsolutePath();
+
+ tablePath = warehouse.concat("/default/").concat(TABLE_NAME);
+ Assert.assertTrue("Should create the table path correctly.", new File(tablePath).mkdirs());
+
+ properties = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name());
+ catalog = new HadoopCatalog(CONF, warehouse);
+
+ env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ env.enableCheckpointing(400);
+ env.setParallelism(parallelism);
+
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inStreamingMode()
Review comment:
That's a great idea, we could reuse almost all of the codes then.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r479832185
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.sink;
+
+import java.util.Locale;
+import java.util.Map;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.util.DataFormatConverters;
+import org.apache.flink.table.runtime.typeutils.RowDataTypeInfo;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.utils.TypeConversions;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.PropertyUtil;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES;
+import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT;
+
+public class FlinkSink {
+
+ private static final String ICEBERG_STREAM_WRITER_NAME = IcebergStreamWriter.class.getSimpleName();
+ private static final String ICEBERG_FILES_COMMITTER_NAME = IcebergFilesCommitter.class.getSimpleName();
+
+ private FlinkSink() {
+ }
+
+ public static Builder forRow(DataStream<Row> input) {
+ return new Builder().forRow(input);
+ }
+
+ public static Builder forRowData(DataStream<RowData> input) {
+ return new Builder().forRowData(input);
+ }
+
+ public static class Builder {
+ private DataStream<Row> rowInput = null;
+ private DataStream<RowData> rowDataInput = null;
+ private TableLoader tableLoader;
+ private Configuration hadoopConf;
+ private Table table;
+ private TableSchema tableSchema;
+
+ private Builder forRow(DataStream<Row> newRowInput) {
+ this.rowInput = newRowInput;
+ return this;
+ }
+
+ private Builder forRowData(DataStream<RowData> newRowDataInput) {
+ this.rowDataInput = newRowDataInput;
+ return this;
+ }
+
+ public Builder table(Table newTable) {
+ this.table = newTable;
+ return this;
+ }
+
+ public Builder tableLoader(TableLoader newTableLoader) {
+ this.tableLoader = newTableLoader;
+ return this;
+ }
+
+ public Builder hadoopConf(Configuration newHadoopConf) {
+ this.hadoopConf = newHadoopConf;
+ return this;
+ }
+
+ public Builder tableSchema(TableSchema newTableSchema) {
+ this.tableSchema = newTableSchema;
+ return this;
+ }
+
+ private DataStream<RowData> convert() {
+ Preconditions.checkArgument(rowInput != null, "The DataStream<Row> to convert shouldn't be null");
+
+ RowType rowType;
+ DataType[] fieldDataTypes;
+ if (tableSchema != null) {
+ rowType = (RowType) tableSchema.toRowDataType().getLogicalType();
+ fieldDataTypes = tableSchema.getFieldDataTypes();
+ } else {
+ rowType = FlinkSchemaUtil.convert(table.schema());
+ fieldDataTypes = TypeConversions.fromLogicalToDataType(rowType.getChildren().toArray(new LogicalType[0]));
+ }
+
+ DataFormatConverters.RowConverter rowConverter = new DataFormatConverters.RowConverter(fieldDataTypes);
+
+ return rowInput.map(rowConverter::toInternal, RowDataTypeInfo.of(rowType));
+ }
+
+ @SuppressWarnings("unchecked")
+ public DataStreamSink<RowData> build() {
+ Preconditions.checkArgument(rowInput != null || rowDataInput != null,
+ "Should initialize input DataStream first with DataStream<Row> or DataStream<RowData>");
+ Preconditions.checkArgument(rowInput == null || rowDataInput == null,
+ "Could only initialize input DataStream with either DataStream<Row> or DataStream<RowData>");
+ Preconditions.checkNotNull(table, "Table shouldn't be null");
+ Preconditions.checkNotNull(tableLoader, "Table loader shouldn't be null");
+ Preconditions.checkNotNull(hadoopConf, "Hadoop configuration shouldn't be null");
+
+ DataStream<RowData> inputStream = rowInput != null ? convert() : rowDataInput;
+
+ IcebergStreamWriter<RowData> streamWriter = createStreamWriter(table, tableSchema);
+ IcebergFilesCommitter filesCommitter = new IcebergFilesCommitter(tableLoader, hadoopConf);
+
+ DataStream<Void> returnStream = inputStream
+ .transform(ICEBERG_STREAM_WRITER_NAME, TypeInformation.of(DataFile.class), streamWriter)
+ .setParallelism(inputStream.getParallelism())
+ .transform(ICEBERG_FILES_COMMITTER_NAME, Types.VOID, filesCommitter)
+ .setParallelism(1)
+ .setMaxParallelism(1);
+
+ return returnStream.addSink(new DiscardingSink())
+ .name(String.format("IcebergSink %s", table.toString()))
+ .setParallelism(1);
Review comment:
Since we are already returning the DataStream, would it make sense to avoid the discarding sink and possibly let people stream the iceberg commit files instead? Like what if I wanted to also feed them into kafka?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482594192
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.Expressions;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.After;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSink extends FlinkCatalogTestBase {
+ private static final String TABLE_NAME = "test_table";
+ private TableEnvironment tEnv;
+ private org.apache.iceberg.Table icebergTable;
+
+ private final FileFormat format;
+ private final boolean isStreamingJob;
+
+ @Parameterized.Parameters(name = "{index}: format={0}, isStreaming={1}, catalogName={2}, baseNamespace={3}")
+ public static Iterable<Object[]> parameters() {
+ List<Object[]> parameters = Lists.newArrayList();
+ for (FileFormat format : new FileFormat[] {FileFormat.ORC, FileFormat.AVRO, FileFormat.PARQUET}) {
+ for (Boolean isStreaming : new Boolean[] {true, false}) {
+ for (Object[] catalogParams : FlinkCatalogTestBase.parameters()) {
+ String catalogName = (String) catalogParams[0];
+ String[] baseNamespace = (String[]) catalogParams[1];
+ parameters.add(new Object[] {format, isStreaming, catalogName, baseNamespace});
+ }
+ }
+ }
+ return parameters;
+ }
+
+ public TestFlinkTableSink(FileFormat format, Boolean isStreamingJob, String catalogName, String[] baseNamespace) {
+ super(catalogName, baseNamespace);
+ this.format = format;
+ this.isStreamingJob = isStreamingJob;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreamingJob) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ env.enableCheckpointing(400);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @Before
+ public void before() {
+ sql("CREATE DATABASE %s", flinkDatabase);
+ sql("USE CATALOG %s", catalogName);
+ sql("USE %s", DATABASE);
+
+ Map<String, String> properties = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name());
+ this.icebergTable = validationCatalog
+ .createTable(TableIdentifier.of(icebergNamespace, TABLE_NAME),
+ SimpleDataUtil.SCHEMA,
+ PartitionSpec.unpartitioned(),
+ properties);
+ }
+
+ @After
+ public void clean() {
+ sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, TABLE_NAME);
+ sql("DROP DATABASE IF EXISTS %s", flinkDatabase);
+ }
+
+ @Test
+ public void testStreamSQL() throws Exception {
+ // Register the rows into a temporary table.
+ Table sourceTable = getTableEnv().fromValues(SimpleDataUtil.FLINK_SCHEMA.toRowDataType(),
+ Expressions.row(1, "hello"),
+ Expressions.row(2, "world"),
+ Expressions.row(3, (String) null),
+ Expressions.row(null, "bar")
+ );
+ getTableEnv().createTemporaryView("sourceTable", sourceTable);
+
+ // Redirect the records from source table to destination table.
+ sql("INSERT INTO %s SELECT id,data from sourceTable", TABLE_NAME);
+
+ // Assert the table records as expected.
+ SimpleDataUtil.assertTableRecords(icebergTable, Lists.newArrayList(
+ SimpleDataUtil.createRecord(1, "hello"),
+ SimpleDataUtil.createRecord(2, "world"),
+ SimpleDataUtil.createRecord(3, null),
+ SimpleDataUtil.createRecord(null, "bar")
+ ));
+ }
+
+ @Test
+ public void testOverwriteTable() throws Exception {
Review comment:
It would be good to also have a partitioned test.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#discussion_r482592976
##########
File path: flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
##########
@@ -126,12 +126,16 @@ public static void assertTableRows(String tablePath, List<RowData> expected) thr
assertTableRecords(tablePath, expectedRecords);
}
- public static void assertTableRecords(String tablePath, List<Record> expected) throws IOException {
- Preconditions.checkArgument(expected != null, "expected records shouldn't be null");
- Table newTable = new HadoopTables().load(tablePath);
- try (CloseableIterable<Record> iterable = IcebergGenerics.read(newTable).build()) {
+ public static void assertTableRecords(Table table, List<Record> expected) throws IOException {
+ table.refresh();
Review comment:
Can this be done automatically when a write completes, or is this a completely separate copy of the table?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#issuecomment-686281454
The broken unit test for jdk11 is ( jdk8 works fine):
```
org.apache.iceberg.spark.sql.TestCreateTableAsSelect > testDataFrameV2Replace[1] FAILED
java.lang.AssertionError: Should have rows matching the source table: number of results should match expected:<3> but was:<6>
at org.junit.Assert.fail(Assert.java:88)
at org.junit.Assert.failNotEquals(Assert.java:834)
at org.junit.Assert.assertEquals(Assert.java:645)
at org.apache.iceberg.spark.SparkTestBase.assertEquals(SparkTestBase.java:100)
at org.apache.iceberg.spark.sql.TestCreateTableAsSelect.testDataFrameV2Replace(TestCreateTableAsSelect.java:206)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1348: Flink: Support table sink.
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1348:
URL: https://github.com/apache/iceberg/pull/1348#issuecomment-686562274
Merged! Thanks for getting this done, @openinx! It's great to see Flink SQL writes working.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org