You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "WencongLiu (via GitHub)" <gi...@apache.org> on 2023/03/20 10:50:50 UTC
[GitHub] [flink] WencongLiu commented on a diff in pull request #21645: [FLINK-30556] Introduce the StaticPartitionFileSplitEnumerator
WencongLiu commented on code in PR #21645:
URL: https://github.com/apache/flink/pull/21645#discussion_r1141937294
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StaticPartitionFileSplitEnumerator.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.flink.connector.file.src.impl;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.connector.file.src.FileSource;
+import org.apache.flink.connector.file.src.FileSourceSplit;
+import org.apache.flink.connector.file.src.PendingSplitsCheckpoint;
+import org.apache.flink.connector.file.src.assigners.FileSplitAssigner;
+import org.apache.flink.connector.file.src.enumerate.FileEnumerator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * An improved SplitEnumerator implementation for bounded / batch {@link FileSource} input. This
+ * enumerator uses the iterator of splits to allocate small batches of FileSourceSplits, which helps
+ * to avoid Java's out-of-memory (OOM) issue. All files present in the configured input directories
+ * are assigned to the readers. Once all files are processed, the source is finished. This class has
+ * a thin implementation. The actual logic for creating the set of FileSourceSplits
+ *
+ * <p>to process, and the logic to decide which reader gets what split, are still in {@link
+ * FileEnumerator} and in {@link FileSplitAssigner}, respectively.
+ */
+@Internal
+public class StaticPartitionFileSplitEnumerator
+ implements SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint<FileSourceSplit>> {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(StaticPartitionFileSplitEnumerator.class);
+
+ private final SplitEnumeratorContext<FileSourceSplit> context;
+
+ private final Iterator<List<FileSourceSplit>> splitsIterator;
+
+ private final FileSplitAssigner.Provider assignerFactory;
+
+ private FileSplitAssigner splitAssigner;
+
+ // ------------------------------------------------------------------------
+
+ public StaticPartitionFileSplitEnumerator(
Review Comment:
I prepare to add a new method `createInputSplitsIterator` with the return type `Iterator<List<FileSourceSplit>>`
in `HiveSourceFileEnumerator`. Then we can modify the overwritten methods `createEnumerator` and `restoreEnumerator` in `HiveSource`. In the overwritten methods, we can invoke`createInputSplitsIterator` to create `StaticPartitionFileSplitEnumerator `.
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StaticPartitionFileSplitEnumerator.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.flink.connector.file.src.impl;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.connector.file.src.FileSource;
+import org.apache.flink.connector.file.src.FileSourceSplit;
+import org.apache.flink.connector.file.src.PendingSplitsCheckpoint;
+import org.apache.flink.connector.file.src.assigners.FileSplitAssigner;
+import org.apache.flink.connector.file.src.enumerate.FileEnumerator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * An improved SplitEnumerator implementation for bounded / batch {@link FileSource} input. This
+ * enumerator uses the iterator of splits to allocate small batches of FileSourceSplits, which helps
+ * to avoid Java's out-of-memory (OOM) issue. All files present in the configured input directories
+ * are assigned to the readers. Once all files are processed, the source is finished. This class has
+ * a thin implementation. The actual logic for creating the set of FileSourceSplits
+ *
+ * <p>to process, and the logic to decide which reader gets what split, are still in {@link
+ * FileEnumerator} and in {@link FileSplitAssigner}, respectively.
+ */
+@Internal
+public class StaticPartitionFileSplitEnumerator
+ implements SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint<FileSourceSplit>> {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(StaticPartitionFileSplitEnumerator.class);
+
+ private final SplitEnumeratorContext<FileSourceSplit> context;
+
+ private final Iterator<List<FileSourceSplit>> splitsIterator;
+
+ private final FileSplitAssigner.Provider assignerFactory;
+
+ private FileSplitAssigner splitAssigner;
+
+ // ------------------------------------------------------------------------
+
+ public StaticPartitionFileSplitEnumerator(
Review Comment:
I prepare to add a new method `createInputSplitsIterator` with the return type `Iterator<List<FileSourceSplit>>`
in `HiveSourceFileEnumerator`. Then we can modify the overwritten methods `createEnumerator` and `restoreEnumerator` in `HiveSource`. In the overwritten methods, we can invoke`createInputSplitsIterator` to create `StaticPartitionFileSplitEnumerator `. @luoyuxia
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org