You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2022/03/11 16:22:09 UTC
[iceberg] branch master updated: Flink: Use a worker pool per job in operators (#4177)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 1b774c2 Flink: Use a worker pool per job in operators (#4177)
1b774c2 is described below
commit 1b774c2c3f054309024e0b0e62688ffa9e25fe20
Author: Yi Tang <ss...@gmail.com>
AuthorDate: Sat Mar 12 00:21:48 2022 +0800
Flink: Use a worker pool per job in operators (#4177)
---
.../java/org/apache/iceberg/util/ThreadPools.java | 22 ++++++++++++------
.../apache/iceberg/flink/FlinkConfigOptions.java | 7 ++++++
.../iceberg/flink/FlinkDynamicTableFactory.java | 2 +-
.../org/apache/iceberg/flink/IcebergTableSink.java | 7 +++++-
.../org/apache/iceberg/flink/sink/FlinkSink.java | 13 ++++++++++-
.../iceberg/flink/sink/IcebergFilesCommitter.java | 27 ++++++++++++++++++----
.../iceberg/flink/source/FlinkInputFormat.java | 7 +++++-
.../apache/iceberg/flink/source/FlinkSource.java | 1 +
.../iceberg/flink/source/FlinkSplitPlanner.java | 18 ++++++++-------
.../apache/iceberg/flink/source/ScanContext.java | 18 +++++++++++++--
.../flink/source/StreamingMonitorFunction.java | 24 ++++++++++++++++++-
.../flink/sink/TestIcebergFilesCommitter.java | 4 +++-
.../apache/iceberg/flink/source/SplitHelpers.java | 4 +++-
.../flink/source/TestStreamingReaderOperator.java | 4 +++-
14 files changed, 129 insertions(+), 29 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/util/ThreadPools.java b/core/src/main/java/org/apache/iceberg/util/ThreadPools.java
index 25367ea..ebf234b 100644
--- a/core/src/main/java/org/apache/iceberg/util/ThreadPools.java
+++ b/core/src/main/java/org/apache/iceberg/util/ThreadPools.java
@@ -38,13 +38,7 @@ public class ThreadPools {
WORKER_THREAD_POOL_SIZE_PROP,
Math.max(2, Runtime.getRuntime().availableProcessors()));
- private static final ExecutorService WORKER_POOL = MoreExecutors.getExitingExecutorService(
- (ThreadPoolExecutor) Executors.newFixedThreadPool(
- WORKER_THREAD_POOL_SIZE,
- new ThreadFactoryBuilder()
- .setDaemon(true)
- .setNameFormat("iceberg-worker-pool-%d")
- .build()));
+ private static final ExecutorService WORKER_POOL = newWorkerPool("iceberg-worker-pool");
/**
* Return an {@link ExecutorService} that uses the "worker" thread-pool.
@@ -61,6 +55,20 @@ public class ThreadPools {
return WORKER_POOL;
}
+ public static ExecutorService newWorkerPool(String namePrefix) {
+ return newWorkerPool(namePrefix, WORKER_THREAD_POOL_SIZE);
+ }
+
+ public static ExecutorService newWorkerPool(String namePrefix, int poolSize) {
+ return MoreExecutors.getExitingExecutorService(
+ (ThreadPoolExecutor) Executors.newFixedThreadPool(
+ poolSize,
+ new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat(namePrefix + "-%d")
+ .build()));
+ }
+
private static int getPoolSize(String systemProperty, int defaultSize) {
String value = System.getProperty(systemProperty);
if (value != null) {
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java
index db5614f..796a709 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableEnvironment;
+import org.apache.iceberg.util.ThreadPools;
/**
* When constructing Flink Iceberg source via Java API,
@@ -74,4 +75,10 @@ public class FlinkConfigOptions {
.intType()
.defaultValue(2048)
.withDescription("The target number of records for Iceberg reader fetch batch.");
+
+ public static final ConfigOption<Integer> TABLE_EXEC_ICEBERG_WORKER_POOL_SIZE =
+ ConfigOptions.key("table.exec.iceberg.worker-pool-size")
+ .intType()
+ .defaultValue(ThreadPools.WORKER_THREAD_POOL_SIZE)
+ .withDescription("The size of workers pool used to plan or scan manifests.");
}
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
index fc0269f..cf6162f 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
@@ -126,7 +126,7 @@ public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, Dynami
objectPath.getObjectName());
}
- return new IcebergTableSink(tableLoader, tableSchema);
+ return new IcebergTableSink(tableLoader, tableSchema, context.getConfiguration());
}
@Override
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java
index 93cf5ad..f2eecb9 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java
@@ -21,6 +21,7 @@ package org.apache.iceberg.flink;
import java.util.List;
import java.util.Map;
+import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.constraints.UniqueConstraint;
import org.apache.flink.table.connector.ChangelogMode;
@@ -36,6 +37,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
public class IcebergTableSink implements DynamicTableSink, SupportsPartitioning, SupportsOverwrite {
private final TableLoader tableLoader;
private final TableSchema tableSchema;
+ private final ReadableConfig readableConfig;
private boolean overwrite = false;
@@ -43,11 +45,13 @@ public class IcebergTableSink implements DynamicTableSink, SupportsPartitioning,
this.tableLoader = toCopy.tableLoader;
this.tableSchema = toCopy.tableSchema;
this.overwrite = toCopy.overwrite;
+ this.readableConfig = toCopy.readableConfig;
}
- public IcebergTableSink(TableLoader tableLoader, TableSchema tableSchema) {
+ public IcebergTableSink(TableLoader tableLoader, TableSchema tableSchema, ReadableConfig readableConfig) {
this.tableLoader = tableLoader;
this.tableSchema = tableSchema;
+ this.readableConfig = readableConfig;
}
@Override
@@ -64,6 +68,7 @@ public class IcebergTableSink implements DynamicTableSink, SupportsPartitioning,
.tableSchema(tableSchema)
.equalityFieldColumns(equalityColumns)
.overwrite(overwrite)
+ .flinkConf(readableConfig)
.append();
}
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
index d15a5f2..407f3a4 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
@@ -29,6 +29,8 @@ import java.util.function.Function;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
@@ -47,6 +49,7 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SerializableTable;
import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.FlinkConfigOptions;
import org.apache.iceberg.flink.FlinkSchemaUtil;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
@@ -136,6 +139,7 @@ public class FlinkSink {
private List<String> equalityFieldColumns = null;
private String uidPrefix = null;
private final Map<String, String> snapshotProperties = Maps.newHashMap();
+ private ReadableConfig readableConfig = new Configuration();
private Builder() {
}
@@ -197,6 +201,11 @@ public class FlinkSink {
return this;
}
+ public Builder flinkConf(ReadableConfig config) {
+ this.readableConfig = config;
+ return this;
+ }
+
/**
* Configure the write {@link DistributionMode} that the flink sink will use. Currently, flink support
* {@link DistributionMode#NONE} and {@link DistributionMode#HASH}.
@@ -381,7 +390,9 @@ public class FlinkSink {
}
private SingleOutputStreamOperator<Void> appendCommitter(SingleOutputStreamOperator<WriteResult> writerStream) {
- IcebergFilesCommitter filesCommitter = new IcebergFilesCommitter(tableLoader, overwrite, snapshotProperties);
+ IcebergFilesCommitter filesCommitter = new IcebergFilesCommitter(
+ tableLoader, overwrite, snapshotProperties,
+ readableConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_WORKER_POOL_SIZE));
SingleOutputStreamOperator<Void> committerStream = writerStream
.transform(operatorName(ICEBERG_FILES_COMMITTER_NAME), Types.VOID, filesCommitter)
.setParallelism(1)
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
index 1dfebb1..bd332d9 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.SortedMap;
+import java.util.concurrent.ExecutorService;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
@@ -55,6 +56,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.ThreadPools;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -110,10 +112,15 @@ class IcebergFilesCommitter extends AbstractStreamOperator<Void>
private static final ListStateDescriptor<SortedMap<Long, byte[]>> STATE_DESCRIPTOR = buildStateDescriptor();
private transient ListState<SortedMap<Long, byte[]>> checkpointsState;
- IcebergFilesCommitter(TableLoader tableLoader, boolean replacePartitions, Map<String, String> snapshotProperties) {
+ private final Integer workerPoolSize;
+ private transient ExecutorService workerPool;
+
+ IcebergFilesCommitter(TableLoader tableLoader, boolean replacePartitions, Map<String, String> snapshotProperties,
+ Integer workerPoolSize) {
this.tableLoader = tableLoader;
this.replacePartitions = replacePartitions;
this.snapshotProperties = snapshotProperties;
+ this.workerPoolSize = workerPoolSize;
}
@Override
@@ -249,7 +256,7 @@ class IcebergFilesCommitter extends AbstractStreamOperator<Void>
Preconditions.checkState(deleteFilesNum == 0, "Cannot overwrite partitions with delete files.");
// Commit the overwrite transaction.
- ReplacePartitions dynamicOverwrite = table.newReplacePartitions();
+ ReplacePartitions dynamicOverwrite = table.newReplacePartitions().scanManifestsWith(workerPool);
int numFiles = 0;
for (WriteResult result : pendingResults.values()) {
@@ -267,7 +274,7 @@ class IcebergFilesCommitter extends AbstractStreamOperator<Void>
if (deleteFilesNum == 0) {
// To be compatible with iceberg format V1.
- AppendFiles appendFiles = table.newAppend();
+ AppendFiles appendFiles = table.newAppend().scanManifestsWith(workerPool);
int numFiles = 0;
for (WriteResult result : pendingResults.values()) {
@@ -292,7 +299,7 @@ class IcebergFilesCommitter extends AbstractStreamOperator<Void>
// rows from data files that are being added in this commit. There is no way for data files added along with
// the delete files to be concurrently removed, so there is no need to validate the files referenced by the
// position delete files that are being committed.
- RowDelta rowDelta = table.newRowDelta();
+ RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
int numDataFiles = result.dataFiles().length;
Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
@@ -351,10 +358,22 @@ class IcebergFilesCommitter extends AbstractStreamOperator<Void>
}
@Override
+ public void open() throws Exception {
+ super.open();
+
+ final String operatorID = getRuntimeContext().getOperatorUniqueID();
+ this.workerPool = ThreadPools.newWorkerPool("iceberg-worker-pool-" + operatorID, workerPoolSize);
+ }
+
+ @Override
public void close() throws Exception {
if (tableLoader != null) {
tableLoader.close();
}
+
+ if (workerPool != null) {
+ workerPool.shutdown();
+ }
}
private static ListStateDescriptor<SortedMap<Long, byte[]>> buildStateDescriptor() {
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java
index 1027b5b..ee71ab7 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java
@@ -20,6 +20,7 @@
package org.apache.iceberg.flink.source;
import java.io.IOException;
+import java.util.concurrent.ExecutorService;
import org.apache.flink.api.common.io.DefaultInputSplitAssigner;
import org.apache.flink.api.common.io.InputFormat;
import org.apache.flink.api.common.io.LocatableInputSplitAssigner;
@@ -34,6 +35,7 @@ import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.util.ThreadPools;
/**
* Flink {@link InputFormat} for Iceberg.
@@ -76,9 +78,12 @@ public class FlinkInputFormat extends RichInputFormat<RowData, FlinkInputSplit>
public FlinkInputSplit[] createInputSplits(int minNumSplits) throws IOException {
// Called in Job manager, so it is OK to load table from catalog.
tableLoader.open();
+ final ExecutorService workerPool = ThreadPools.newWorkerPool("iceberg-plan-worker-pool", context.planParallelism());
try (TableLoader loader = tableLoader) {
Table table = loader.loadTable();
- return FlinkSplitPlanner.planInputSplits(table, context);
+ return FlinkSplitPlanner.planInputSplits(table, context, workerPool);
+ } finally {
+ workerPool.shutdown();
}
}
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
index df8009f..336b2ae 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
@@ -212,6 +212,7 @@ public class FlinkSource {
contextBuilder.project(FlinkSchemaUtil.convert(icebergSchema, projectedSchema));
}
contextBuilder.exposeLocality(localityEnabled());
+ contextBuilder.planParallelism(readableConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_WORKER_POOL_SIZE));
return new FlinkInputFormat(tableLoader, icebergSchema, io, encryption, contextBuilder.build());
}
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSplitPlanner.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSplitPlanner.java
index 213f9c9..10e931d 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSplitPlanner.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSplitPlanner.java
@@ -22,6 +22,7 @@ package org.apache.iceberg.flink.source;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.List;
+import java.util.concurrent.ExecutorService;
import org.apache.flink.annotation.Internal;
import org.apache.iceberg.CombinedScanTask;
import org.apache.iceberg.Table;
@@ -33,22 +34,21 @@ import org.apache.iceberg.hadoop.Util;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.util.Tasks;
-import org.apache.iceberg.util.ThreadPools;
@Internal
public class FlinkSplitPlanner {
private FlinkSplitPlanner() {
}
- static FlinkInputSplit[] planInputSplits(Table table, ScanContext context) {
- try (CloseableIterable<CombinedScanTask> tasksIterable = planTasks(table, context)) {
+ static FlinkInputSplit[] planInputSplits(Table table, ScanContext context, ExecutorService workerPool) {
+ try (CloseableIterable<CombinedScanTask> tasksIterable = planTasks(table, context, workerPool)) {
List<CombinedScanTask> tasks = Lists.newArrayList(tasksIterable);
FlinkInputSplit[] splits = new FlinkInputSplit[tasks.size()];
boolean exposeLocality = context.exposeLocality();
Tasks.range(tasks.size())
.stopOnFailure()
- .executeWith(exposeLocality ? ThreadPools.getWorkerPool() : null)
+ .executeWith(exposeLocality ? workerPool : null)
.run(index -> {
CombinedScanTask task = tasks.get(index);
String[] hostnames = null;
@@ -66,8 +66,9 @@ public class FlinkSplitPlanner {
/**
* This returns splits for the FLIP-27 source
*/
- public static List<IcebergSourceSplit> planIcebergSourceSplits(Table table, ScanContext context) {
- try (CloseableIterable<CombinedScanTask> tasksIterable = planTasks(table, context)) {
+ public static List<IcebergSourceSplit> planIcebergSourceSplits(
+ Table table, ScanContext context, ExecutorService workerPool) {
+ try (CloseableIterable<CombinedScanTask> tasksIterable = planTasks(table, context, workerPool)) {
return Lists.newArrayList(CloseableIterable.transform(tasksIterable,
task -> IcebergSourceSplit.fromCombinedScanTask(task)));
} catch (IOException e) {
@@ -75,11 +76,12 @@ public class FlinkSplitPlanner {
}
}
- static CloseableIterable<CombinedScanTask> planTasks(Table table, ScanContext context) {
+ static CloseableIterable<CombinedScanTask> planTasks(Table table, ScanContext context, ExecutorService workerPool) {
TableScan scan = table
.newScan()
.caseSensitive(context.caseSensitive())
- .project(context.project());
+ .project(context.project())
+ .planWith(workerPool);
if (context.includeColumnStats()) {
scan = scan.includeColumnStats();
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
index 25b3bd0..6c9ed11 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
@@ -28,6 +28,7 @@ import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.iceberg.Schema;
import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.flink.FlinkConfigOptions;
import static org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING;
@@ -88,11 +89,13 @@ class ScanContext implements Serializable {
private final List<Expression> filters;
private final long limit;
private final boolean includeColumnStats;
+ private final Integer planParallelism;
private ScanContext(boolean caseSensitive, Long snapshotId, Long startSnapshotId, Long endSnapshotId,
Long asOfTimestamp, Long splitSize, Integer splitLookback, Long splitOpenFileCost,
boolean isStreaming, Duration monitorInterval, String nameMapping, Schema schema,
- List<Expression> filters, long limit, boolean includeColumnStats, boolean exposeLocality) {
+ List<Expression> filters, long limit, boolean includeColumnStats, boolean exposeLocality,
+ Integer planParallelism) {
this.caseSensitive = caseSensitive;
this.snapshotId = snapshotId;
this.startSnapshotId = startSnapshotId;
@@ -110,6 +113,7 @@ class ScanContext implements Serializable {
this.limit = limit;
this.includeColumnStats = includeColumnStats;
this.exposeLocality = exposeLocality;
+ this.planParallelism = planParallelism;
}
boolean caseSensitive() {
@@ -176,6 +180,10 @@ class ScanContext implements Serializable {
return exposeLocality;
}
+ Integer planParallelism() {
+ return planParallelism;
+ }
+
ScanContext copyWithAppendsBetween(long newStartSnapshotId, long newEndSnapshotId) {
return ScanContext.builder()
.caseSensitive(caseSensitive)
@@ -239,6 +247,7 @@ class ScanContext implements Serializable {
private long limit = -1L;
private boolean includeColumnStats = INCLUDE_COLUMN_STATS.defaultValue();
private boolean exposeLocality;
+ private Integer planParallelism = FlinkConfigOptions.TABLE_EXEC_ICEBERG_WORKER_POOL_SIZE.defaultValue();
private Builder() {
}
@@ -323,6 +332,11 @@ class ScanContext implements Serializable {
return this;
}
+ Builder planParallelism(Integer parallelism) {
+ this.planParallelism = parallelism;
+ return this;
+ }
+
Builder fromProperties(Map<String, String> properties) {
Configuration config = new Configuration();
properties.forEach(config::setString);
@@ -345,7 +359,7 @@ class ScanContext implements Serializable {
return new ScanContext(caseSensitive, snapshotId, startSnapshotId,
endSnapshotId, asOfTimestamp, splitSize, splitLookback,
splitOpenFileCost, isStreaming, monitorInterval, nameMapping, projectedSchema,
- filters, limit, includeColumnStats, exposeLocality);
+ filters, limit, includeColumnStats, exposeLocality, planParallelism);
}
}
}
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/StreamingMonitorFunction.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/StreamingMonitorFunction.java
index 8bfad6d..d697009 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/StreamingMonitorFunction.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/StreamingMonitorFunction.java
@@ -21,18 +21,24 @@ package org.apache.iceberg.flink.source;
import java.io.IOException;
import java.io.UncheckedIOException;
+import java.util.concurrent.ExecutorService;
+import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
+import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.ThreadPools;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,6 +73,7 @@ public class StreamingMonitorFunction extends RichSourceFunction<FlinkInputSplit
private transient SourceContext<FlinkInputSplit> sourceContext;
private transient Table table;
private transient ListState<Long> lastSnapshotIdState;
+ private transient ExecutorService workerPool;
public StreamingMonitorFunction(TableLoader tableLoader, ScanContext scanContext) {
Preconditions.checkArgument(scanContext.snapshotId() == null,
@@ -80,6 +87,17 @@ public class StreamingMonitorFunction extends RichSourceFunction<FlinkInputSplit
}
@Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+
+ final RuntimeContext runtimeContext = getRuntimeContext();
+ ValidationException.check(
+ runtimeContext instanceof StreamingRuntimeContext, "context should be instance of StreamingRuntimeContext");
+ final String operatorID = ((StreamingRuntimeContext) runtimeContext).getOperatorUniqueID();
+ this.workerPool = ThreadPools.newWorkerPool("iceberg-worker-pool-" + operatorID, scanContext.planParallelism());
+ }
+
+ @Override
public void initializeState(FunctionInitializationContext context) throws Exception {
// Load iceberg table from table loader.
tableLoader.open();
@@ -140,7 +158,7 @@ public class StreamingMonitorFunction extends RichSourceFunction<FlinkInputSplit
newScanContext = scanContext.copyWithAppendsBetween(lastSnapshotId, snapshotId);
}
- FlinkInputSplit[] splits = FlinkSplitPlanner.planInputSplits(table, newScanContext);
+ FlinkInputSplit[] splits = FlinkSplitPlanner.planInputSplits(table, newScanContext, workerPool);
for (FlinkInputSplit split : splits) {
sourceContext.collect(split);
}
@@ -173,5 +191,9 @@ public class StreamingMonitorFunction extends RichSourceFunction<FlinkInputSplit
@Override
public void close() {
cancel();
+
+ if (workerPool != null) {
+ workerPool.shutdown();
+ }
}
}
diff --git a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
index 73a17aa..78ffe72 100644
--- a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
+++ b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
@@ -58,6 +58,7 @@ import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.util.Pair;
+import org.apache.iceberg.util.ThreadPools;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
@@ -826,7 +827,8 @@ public class TestIcebergFilesCommitter extends TableTestBase {
@SuppressWarnings("unchecked")
public <T extends StreamOperator<Void>> T createStreamOperator(StreamOperatorParameters<Void> param) {
IcebergFilesCommitter committer = new IcebergFilesCommitter(new TestTableLoader(tablePath), false,
- Collections.singletonMap("flink.test", TestIcebergFilesCommitter.class.getName()));
+ Collections.singletonMap("flink.test", TestIcebergFilesCommitter.class.getName()),
+ ThreadPools.WORKER_THREAD_POOL_SIZE);
committer.setup(param.getContainingTask(), param.getStreamConfig(), param.getOutput());
return (T) committer;
}
diff --git a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/SplitHelpers.java b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/SplitHelpers.java
index df98808..bc302fa 100644
--- a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/SplitHelpers.java
+++ b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/SplitHelpers.java
@@ -35,6 +35,7 @@ import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.flink.source.split.IcebergSourceSplit;
import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.ThreadPools;
import org.junit.Assert;
import org.junit.rules.TemporaryFolder;
@@ -72,7 +73,8 @@ public class SplitHelpers {
}
final ScanContext scanContext = ScanContext.builder().build();
- final List<IcebergSourceSplit> splits = FlinkSplitPlanner.planIcebergSourceSplits(table, scanContext);
+ final List<IcebergSourceSplit> splits = FlinkSplitPlanner.planIcebergSourceSplits(
+ table, scanContext, ThreadPools.getWorkerPool());
return splits.stream()
.flatMap(split -> {
List<List<FileScanTask>> filesList = Lists.partition(
diff --git a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/TestStreamingReaderOperator.java b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/TestStreamingReaderOperator.java
index 7978af6..0c31911 100644
--- a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/TestStreamingReaderOperator.java
+++ b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/source/TestStreamingReaderOperator.java
@@ -46,6 +46,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.ThreadPools;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -254,7 +255,8 @@ public class TestStreamingReaderOperator extends TableTestBase {
.build();
}
- Collections.addAll(inputSplits, FlinkSplitPlanner.planInputSplits(table, scanContext));
+ Collections.addAll(inputSplits, FlinkSplitPlanner.planInputSplits(
+ table, scanContext, ThreadPools.getWorkerPool()));
}
return inputSplits;