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/04/28 06:52:31 UTC
[GitHub] [incubator-iceberg] jerryshao opened a new pull request #796: Support Spark Structured Streaming Read for Iceberg
jerryshao opened a new pull request #796:
URL: https://github.com/apache/incubator-iceberg/pull/796
This patch adds support for Spark Structured Streaming Read of Iceberg table [#179]. The key implementation is based on incremental scan, and using number of tasks per batch to control the size of each batch. The basic idea is similar to delta lake, and current implementation only tracks added files for each snapshot based on incremental scan.
Currently this patch can be worked with Spark 2.4 smoothly, and with UT added in key code path. Here I would like to bring this initial version out to see if this is a feasible solution, what else should be changed to improve it? In the meantime I will add more UTs and do the code refactor works.
----------------------------------------------------------------
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] [incubator-iceberg] rdblue commented on a change in pull request #796: Support Spark Structured Streaming Read for Iceberg
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #796:
URL: https://github.com/apache/incubator-iceberg/pull/796#discussion_r418784273
##########
File path: api/src/main/java/org/apache/iceberg/TableScan.java
##########
@@ -165,6 +166,22 @@ default TableScan select(String... columns) {
*/
CloseableIterable<FileScanTask> planFiles();
+ /**
+ * Plan the {@link FileScanTask files} for the specified {@link ManifestFile manifests} that will be read by this
+ * scan.
+ * <p>
+ * Each file has a residual expression that should be applied to filter the file's rows.
+ * <p>
+ * This simple plan returns file scans for each file from position 0 to the file's length. For
+ * planning that will combine small files, split large files, and attempt to balance work, use
+ * {@link #planTasks()} instead.
+ *
+ * @param manifests the specified manifests for this scan.
+ * @param filterAddedFiles whether to filter just added files or all files.
+ * @return an Iterable of file tasks that are required by this scan
+ */
+ CloseableIterable<FileScanTask> planFilesForManifests(List<ManifestFile> manifests, boolean filterAddedFiles);
Review comment:
This is something that a `ManifestGroup` should be used for instead. I don't think we want to add a new method to `TableScan`.
----------------------------------------------------------------
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] [incubator-iceberg] jerryshao commented on a change in pull request #796: Support Spark Structured Streaming Read for Iceberg
Posted by GitBox <gi...@apache.org>.
jerryshao commented on a change in pull request #796:
URL: https://github.com/apache/incubator-iceberg/pull/796#discussion_r420521123
##########
File path: api/src/main/java/org/apache/iceberg/TableScan.java
##########
@@ -165,6 +166,22 @@ default TableScan select(String... columns) {
*/
CloseableIterable<FileScanTask> planFiles();
+ /**
+ * Plan the {@link FileScanTask files} for the specified {@link ManifestFile manifests} that will be read by this
+ * scan.
+ * <p>
+ * Each file has a residual expression that should be applied to filter the file's rows.
+ * <p>
+ * This simple plan returns file scans for each file from position 0 to the file's length. For
+ * planning that will combine small files, split large files, and attempt to balance work, use
+ * {@link #planTasks()} instead.
+ *
+ * @param manifests the specified manifests for this scan.
+ * @param filterAddedFiles whether to filter just added files or all files.
+ * @return an Iterable of file tasks that are required by this scan
+ */
+ CloseableIterable<FileScanTask> planFilesForManifests(List<ManifestFile> manifests, boolean filterAddedFiles);
Review comment:
Sorry for the late response @rdblue , I will take a look at this implementation today.
----------------------------------------------------------------
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] [incubator-iceberg] rdblue commented on pull request #796: Support Spark Structured Streaming Read for Iceberg
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #796:
URL: https://github.com/apache/incubator-iceberg/pull/796#issuecomment-619279012
`PackingIterable` is not a `CloseableIterable`, so if you use it to wrap a `CloseableIterable` and don't handle close, it would lead to the input stream leak.
----------------------------------------------------------------
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] [incubator-iceberg] rdblue commented on pull request #796: Support Spark Structured Streaming Read for Iceberg
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #796:
URL: https://github.com/apache/incubator-iceberg/pull/796#issuecomment-619279054
I'll try to take a look this weekend. Thanks @jerryshao!
----------------------------------------------------------------
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] [incubator-iceberg] rdblue commented on pull request #796: Support Spark Structured Streaming Read for Iceberg
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #796:
URL: https://github.com/apache/incubator-iceberg/pull/796#issuecomment-620930813
Sorry for the delay getting to this. I was working on the release, mostly. I should have time soon.
----------------------------------------------------------------
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] [incubator-iceberg] rdblue commented on a change in pull request #796: Support Spark Structured Streaming Read for Iceberg
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #796:
URL: https://github.com/apache/incubator-iceberg/pull/796#discussion_r418807378
##########
File path: api/src/main/java/org/apache/iceberg/TableScan.java
##########
@@ -165,6 +166,22 @@ default TableScan select(String... columns) {
*/
CloseableIterable<FileScanTask> planFiles();
+ /**
+ * Plan the {@link FileScanTask files} for the specified {@link ManifestFile manifests} that will be read by this
+ * scan.
+ * <p>
+ * Each file has a residual expression that should be applied to filter the file's rows.
+ * <p>
+ * This simple plan returns file scans for each file from position 0 to the file's length. For
+ * planning that will combine small files, split large files, and attempt to balance work, use
+ * {@link #planTasks()} instead.
+ *
+ * @param manifests the specified manifests for this scan.
+ * @param filterAddedFiles whether to filter just added files or all files.
+ * @return an Iterable of file tasks that are required by this scan
+ */
+ CloseableIterable<FileScanTask> planFilesForManifests(List<ManifestFile> manifests, boolean filterAddedFiles);
Review comment:
How about removing the changes to scans and using the same underlying planning instead?
Here's a class that uses `ManifestGroup` just like the incremental scans. You can use it to get a `CloseableIterable` of `FileScanTask` for a single Snapshot, either full or just appends:
```java
package org.apache.iceberg;
import com.google.common.collect.Iterables;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.FileIO;
class Snapshots {
public IterableBuilder from(Snapshot snapshot, FileIO io) {
return new IterableBuilder(io, snapshot);
}
public class IterableBuilder {
private final FileIO io;
private final Snapshot snapshot;
private Expression rowFilter;
private IterableBuilder(FileIO io, Snapshot snapshot) {
this.io = io;
this.snapshot = snapshot;
}
public IterableBuilder filter(Expression newRowFilter) {
this.rowFilter = newRowFilter;
return this;
}
public CloseableIterable<FileScanTask> full() {
return new ManifestGroup(io, snapshot.manifests())
.filterData(rowFilter)
.planFiles();
}
public CloseableIterable<FileScanTask> appends() {
Iterable<ManifestFile> newManifests = Iterables.filter(snapshot.manifests(),
manifest -> manifest.snapshotId() == snapshot.snapshotId());
return new ManifestGroup(io, newManifests)
.filterData(rowFilter)
.filterManifestEntries(entry -> entry.snapshotId() == snapshot.snapshotId())
.ignoreDeleted()
.ignoreExisting()
.planFiles();
}
}
}
```
The advantage here is that by calling `ManifestGroup` directly, you don't need to change how scans work at all -- so no need to change how the threading works, although that would be a great independent PR to open.
This gives you deterministic access to the files you need from a snapshot. For a starting snapshot, you'd get a `full()` task iterable and use the offset to filter. For other snapshots, you'd get the `appends()` task iterable and do the same offset filtering. I think this will make the implementation a lot more compact.
----------------------------------------------------------------
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