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 2021/02/19 22:30:12 UTC
[GitHub] [iceberg] aokolnychyi opened a new pull request #2256: Spark: Refactor BaseSparkAction
aokolnychyi opened a new pull request #2256:
URL: https://github.com/apache/iceberg/pull/2256
This PR refactors `BaseSparkAction` and prepares it for upcoming 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] aokolnychyi commented on a change in pull request #2256: Spark: Refactor BaseSparkAction
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2256:
URL: https://github.com/apache/iceberg/pull/2256#discussion_r579515625
##########
File path: spark/src/main/java/org/apache/iceberg/actions/BaseSparkAction.java
##########
@@ -101,47 +117,43 @@
return allManifests.flatMap(new ReadManifest(ioBroadcast), Encoders.STRING()).toDF("file_path");
}
- protected Dataset<Row> buildManifestFileDF(SparkSession spark, String tableName) {
- return loadMetadataTable(spark, tableName, table().location(), ALL_MANIFESTS).selectExpr("path as file_path");
+ protected Dataset<Row> buildManifestFileDF(Table table) {
+ return loadMetadataTable(table, ALL_MANIFESTS).selectExpr("path as file_path");
}
- protected Dataset<Row> buildManifestListDF(SparkSession spark, Table table) {
+ protected Dataset<Row> buildManifestListDF(Table table) {
List<String> manifestLists = getManifestListPaths(table.snapshots());
return spark.createDataset(manifestLists, Encoders.STRING()).toDF("file_path");
}
- protected Dataset<Row> buildManifestListDF(SparkSession spark, String metadataFileLocation) {
- StaticTableOperations ops = new StaticTableOperations(metadataFileLocation, table().io());
- return buildManifestListDF(spark, new BaseTable(ops, table().name()));
- }
-
- protected Dataset<Row> buildOtherMetadataFileDF(SparkSession spark, TableOperations ops) {
+ protected Dataset<Row> buildOtherMetadataFileDF(TableOperations ops) {
List<String> otherMetadataFiles = getOtherMetadataFilePaths(ops);
return spark.createDataset(otherMetadataFiles, Encoders.STRING()).toDF("file_path");
}
- protected Dataset<Row> buildValidMetadataFileDF(SparkSession spark, Table table, TableOperations ops) {
- Dataset<Row> manifestDF = buildManifestFileDF(spark, table.name());
- Dataset<Row> manifestListDF = buildManifestListDF(spark, table);
- Dataset<Row> otherMetadataFileDF = buildOtherMetadataFileDF(spark, ops);
+ protected Dataset<Row> buildValidMetadataFileDF(Table table, TableOperations ops) {
+ Dataset<Row> manifestDF = buildManifestFileDF(table);
+ Dataset<Row> manifestListDF = buildManifestListDF(table);
+ Dataset<Row> otherMetadataFileDF = buildOtherMetadataFileDF(ops);
return manifestDF.union(otherMetadataFileDF).union(manifestListDF);
}
// Attempt to use Spark3 Catalog resolution if available on the path
private static final DynMethods.UnboundMethod LOAD_CATALOG = DynMethods.builder("loadCatalogMetadataTable")
- .hiddenImpl("org.apache.iceberg.spark.Spark3Util",
- SparkSession.class, String.class, MetadataTableType.class)
+ .hiddenImpl("org.apache.iceberg.spark.Spark3Util", SparkSession.class, String.class, MetadataTableType.class)
.orNoop()
.build();
- private static Dataset<Row> loadCatalogMetadataTable(SparkSession spark, String tableName, MetadataTableType type) {
Review comment:
This does not have to be static anymore.
----------------------------------------------------------------
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] aokolnychyi commented on pull request #2256: Spark: Refactor BaseSparkAction
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2256:
URL: https://github.com/apache/iceberg/pull/2256#issuecomment-782429034
This one actually depends on #2255. I'll reopen later.
----------------------------------------------------------------
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] aokolnychyi commented on a change in pull request #2256: Spark: Refactor BaseSparkAction
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2256:
URL: https://github.com/apache/iceberg/pull/2256#discussion_r579516592
##########
File path: spark/src/main/java/org/apache/iceberg/actions/BaseSparkAction.java
##########
@@ -101,47 +117,43 @@
return allManifests.flatMap(new ReadManifest(ioBroadcast), Encoders.STRING()).toDF("file_path");
}
- protected Dataset<Row> buildManifestFileDF(SparkSession spark, String tableName) {
- return loadMetadataTable(spark, tableName, table().location(), ALL_MANIFESTS).selectExpr("path as file_path");
+ protected Dataset<Row> buildManifestFileDF(Table table) {
+ return loadMetadataTable(table, ALL_MANIFESTS).selectExpr("path as file_path");
}
- protected Dataset<Row> buildManifestListDF(SparkSession spark, Table table) {
+ protected Dataset<Row> buildManifestListDF(Table table) {
List<String> manifestLists = getManifestListPaths(table.snapshots());
return spark.createDataset(manifestLists, Encoders.STRING()).toDF("file_path");
}
- protected Dataset<Row> buildManifestListDF(SparkSession spark, String metadataFileLocation) {
- StaticTableOperations ops = new StaticTableOperations(metadataFileLocation, table().io());
- return buildManifestListDF(spark, new BaseTable(ops, table().name()));
- }
-
- protected Dataset<Row> buildOtherMetadataFileDF(SparkSession spark, TableOperations ops) {
+ protected Dataset<Row> buildOtherMetadataFileDF(TableOperations ops) {
List<String> otherMetadataFiles = getOtherMetadataFilePaths(ops);
return spark.createDataset(otherMetadataFiles, Encoders.STRING()).toDF("file_path");
}
- protected Dataset<Row> buildValidMetadataFileDF(SparkSession spark, Table table, TableOperations ops) {
- Dataset<Row> manifestDF = buildManifestFileDF(spark, table.name());
- Dataset<Row> manifestListDF = buildManifestListDF(spark, table);
- Dataset<Row> otherMetadataFileDF = buildOtherMetadataFileDF(spark, ops);
+ protected Dataset<Row> buildValidMetadataFileDF(Table table, TableOperations ops) {
+ Dataset<Row> manifestDF = buildManifestFileDF(table);
+ Dataset<Row> manifestListDF = buildManifestListDF(table);
+ Dataset<Row> otherMetadataFileDF = buildOtherMetadataFileDF(ops);
return manifestDF.union(otherMetadataFileDF).union(manifestListDF);
}
// Attempt to use Spark3 Catalog resolution if available on the path
private static final DynMethods.UnboundMethod LOAD_CATALOG = DynMethods.builder("loadCatalogMetadataTable")
- .hiddenImpl("org.apache.iceberg.spark.Spark3Util",
- SparkSession.class, String.class, MetadataTableType.class)
+ .hiddenImpl("org.apache.iceberg.spark.Spark3Util", SparkSession.class, String.class, MetadataTableType.class)
.orNoop()
.build();
- private static Dataset<Row> loadCatalogMetadataTable(SparkSession spark, String tableName, MetadataTableType type) {
+ private Dataset<Row> loadCatalogMetadataTable(String tableName, MetadataTableType type) {
Preconditions.checkArgument(!LOAD_CATALOG.isNoop(), "Cannot find Spark3Util class but Spark3 is in use");
return LOAD_CATALOG.asStatic().invoke(spark, tableName, type);
}
- protected static Dataset<Row> loadMetadataTable(SparkSession spark, String tableName, String tableLocation,
Review comment:
Previously, there was an inconsistency in what values we passed to this method. For example, we could pass `tableName` as the metadata location but table location as the root table location.
----------------------------------------------------------------
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] aokolnychyi closed pull request #2256: Spark: Refactor BaseSparkAction
Posted by GitBox <gi...@apache.org>.
aokolnychyi closed pull request #2256:
URL: https://github.com/apache/iceberg/pull/2256
----------------------------------------------------------------
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] aokolnychyi commented on a change in pull request #2256: Spark: Refactor BaseSparkAction
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2256:
URL: https://github.com/apache/iceberg/pull/2256#discussion_r579516592
##########
File path: spark/src/main/java/org/apache/iceberg/actions/BaseSparkAction.java
##########
@@ -101,47 +117,43 @@
return allManifests.flatMap(new ReadManifest(ioBroadcast), Encoders.STRING()).toDF("file_path");
}
- protected Dataset<Row> buildManifestFileDF(SparkSession spark, String tableName) {
- return loadMetadataTable(spark, tableName, table().location(), ALL_MANIFESTS).selectExpr("path as file_path");
+ protected Dataset<Row> buildManifestFileDF(Table table) {
+ return loadMetadataTable(table, ALL_MANIFESTS).selectExpr("path as file_path");
}
- protected Dataset<Row> buildManifestListDF(SparkSession spark, Table table) {
+ protected Dataset<Row> buildManifestListDF(Table table) {
List<String> manifestLists = getManifestListPaths(table.snapshots());
return spark.createDataset(manifestLists, Encoders.STRING()).toDF("file_path");
}
- protected Dataset<Row> buildManifestListDF(SparkSession spark, String metadataFileLocation) {
- StaticTableOperations ops = new StaticTableOperations(metadataFileLocation, table().io());
- return buildManifestListDF(spark, new BaseTable(ops, table().name()));
- }
-
- protected Dataset<Row> buildOtherMetadataFileDF(SparkSession spark, TableOperations ops) {
+ protected Dataset<Row> buildOtherMetadataFileDF(TableOperations ops) {
List<String> otherMetadataFiles = getOtherMetadataFilePaths(ops);
return spark.createDataset(otherMetadataFiles, Encoders.STRING()).toDF("file_path");
}
- protected Dataset<Row> buildValidMetadataFileDF(SparkSession spark, Table table, TableOperations ops) {
- Dataset<Row> manifestDF = buildManifestFileDF(spark, table.name());
- Dataset<Row> manifestListDF = buildManifestListDF(spark, table);
- Dataset<Row> otherMetadataFileDF = buildOtherMetadataFileDF(spark, ops);
+ protected Dataset<Row> buildValidMetadataFileDF(Table table, TableOperations ops) {
+ Dataset<Row> manifestDF = buildManifestFileDF(table);
+ Dataset<Row> manifestListDF = buildManifestListDF(table);
+ Dataset<Row> otherMetadataFileDF = buildOtherMetadataFileDF(ops);
return manifestDF.union(otherMetadataFileDF).union(manifestListDF);
}
// Attempt to use Spark3 Catalog resolution if available on the path
private static final DynMethods.UnboundMethod LOAD_CATALOG = DynMethods.builder("loadCatalogMetadataTable")
- .hiddenImpl("org.apache.iceberg.spark.Spark3Util",
- SparkSession.class, String.class, MetadataTableType.class)
+ .hiddenImpl("org.apache.iceberg.spark.Spark3Util", SparkSession.class, String.class, MetadataTableType.class)
.orNoop()
.build();
- private static Dataset<Row> loadCatalogMetadataTable(SparkSession spark, String tableName, MetadataTableType type) {
+ private Dataset<Row> loadCatalogMetadataTable(String tableName, MetadataTableType type) {
Preconditions.checkArgument(!LOAD_CATALOG.isNoop(), "Cannot find Spark3Util class but Spark3 is in use");
return LOAD_CATALOG.asStatic().invoke(spark, tableName, type);
}
- protected static Dataset<Row> loadMetadataTable(SparkSession spark, String tableName, String tableLocation,
Review comment:
Previously, there was an inconsistency in what values we passed to this method.
For example, we could pass `tableName` as the metadata location but table location as the root table location. I moved it to use the `Table` object directly to avoid any surprises.
----------------------------------------------------------------
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] aokolnychyi commented on a change in pull request #2256: Spark: Refactor BaseSparkAction
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2256:
URL: https://github.com/apache/iceberg/pull/2256#discussion_r579515842
##########
File path: spark/src/main/java/org/apache/iceberg/actions/BaseSparkAction.java
##########
@@ -84,15 +98,17 @@
return otherMetadataFiles;
}
- protected Dataset<Row> buildValidDataFileDF(SparkSession spark) {
Review comment:
I removed `SparkSession` from args in all these methods.
----------------------------------------------------------------
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] aokolnychyi commented on a change in pull request #2256: Spark: Refactor BaseSparkAction
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2256:
URL: https://github.com/apache/iceberg/pull/2256#discussion_r579515420
##########
File path: spark/src/main/java/org/apache/iceberg/actions/BaseSparkAction.java
##########
@@ -48,7 +48,21 @@
abstract class BaseSparkAction<R> implements Action<R> {
- protected abstract Table table();
+ private final SparkSession spark;
Review comment:
Previously, we always assumed there is going to be a table we operate on. That's not the case with actions like snapshot and migrate. Plus, it does not make sense to init the Java context in all actions. We can do this in parent.
----------------------------------------------------------------
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