Skip to content

Commit 603df6a

Browse files
juripetersenjuripetersen
authored
Merge pull request #656 from ChristofferEmilKristensen/main
Support for Source and Sink Operators for Apache Iceberg for the Java Platform
2 parents 7337a78 + 528f563 commit 603df6a

12 files changed

Lines changed: 1053 additions & 4 deletions

File tree

‎wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuanta.scala‎

Lines changed: 44 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,11 @@ import scala.collection.JavaConversions
4747
import scala.collection.JavaConversions._
4848
import scala.reflect._
4949

50+
import org.apache.iceberg.Schema
51+
import org.apache.iceberg.FileFormat
52+
import org.apache.iceberg.catalog.{Catalog, TableIdentifier}
53+
54+
5055
/**
5156
* Represents an intermediate result/data flow edge in a [[WayangPlan]].
5257
*
@@ -1013,7 +1018,6 @@ class DataQuanta[Out: ClassTag](val operator: ElementaryOperator, outputIndex: I
10131018
this.planBuilder.buildAndExplain(toJson)
10141019
}
10151020

1016-
10171021
/**
10181022
* Write the data quanta in this instance to a text file. Triggers execution.
10191023
*
@@ -1027,6 +1031,44 @@ class DataQuanta[Out: ClassTag](val operator: ElementaryOperator, outputIndex: I
10271031
writeTextFileJava(url, toSerializableFunction(formatterUdf), udfLoad)
10281032
}
10291033

1034+
/**
1035+
* Write the data quanta in this instance to a iceberg table. Triggers execution.
1036+
*
1037+
* @param catalog Iceberg Catalog
1038+
* @param schema Iceberg Schema of the table to create
1039+
* @param tableIdentifier Iceberg Table Identifier of the table to create
1040+
* @param outputFileFormat File format of the output data files
1041+
*/
1042+
1043+
def writeIcebergTable(catalog: Catalog,
1044+
schema: Schema,
1045+
tableIdentifier: TableIdentifier,
1046+
outputFileFormat: FileFormat): Unit = {
1047+
writeIcebergTableJava(catalog, schema, tableIdentifier, outputFileFormat)
1048+
}
1049+
1050+
/**
1051+
* Write the data quanta in this instance to a iceberg table. Triggers execution.
1052+
*
1053+
* @param catalog Iceberg Catalog
1054+
* @param schema Iceberg Schema of the table to create
1055+
* @param tableIdentifier Iceberg Table Identifier of the table to create
1056+
* @param outputFileFormat File format of the output data files
1057+
*/
1058+
def writeIcebergTableJava(
1059+
catalog: Catalog,
1060+
schema: Schema,
1061+
tableIdentifier: TableIdentifier,
1062+
outputFileFormat: FileFormat ): Unit = {
1063+
1064+
val sink = new ApacheIcebergSink(catalog, schema, tableIdentifier, outputFileFormat)
1065+
1066+
sink.setName(s"*#-> Write to Iceberg Table Sink ")
1067+
this.connectTo(sink, 0)
1068+
this.planBuilder.sinks += sink
1069+
this.planBuilder.buildAndExecute()
1070+
this.planBuilder.sinks.clear()
1071+
}
10301072
def writeParquet(url: String,
10311073
overwrite: Boolean = false,
10321074
preferDataset: Boolean = false)(implicit ev: Out =:= Record): Unit =
@@ -1095,6 +1137,7 @@ class DataQuanta[Out: ClassTag](val operator: ElementaryOperator, outputIndex: I
10951137
this.planBuilder.sinks.clear()
10961138
}
10971139

1140+
10981141
private def writeParquetJava(url: String, overwrite: Boolean, preferDataset: Boolean)(implicit ev: Out =:= Record): Unit = {
10991142
val _ = ev
11001143
val sink = new ParquetSink(url, overwrite, preferDataset)

‎wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuantaBuilder.scala‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,10 @@ import org.apache.wayang.core.types.DataSetType
4040
import org.apache.wayang.core.util.{Logging, ReflectionUtils, WayangCollections, Tuple => WayangTuple}
4141
import org.apache.wayang.core.plan.wayangplan.OutputSlot
4242

43+
import org.apache.iceberg.Schema
44+
import org.apache.iceberg.FileFormat
45+
import org.apache.iceberg.catalog.{Catalog, TableIdentifier}
46+
4347

4448

4549
import scala.collection.mutable.ListBuffer
@@ -499,6 +503,27 @@ trait DataQuantaBuilder[+This <: DataQuantaBuilder[_, Out], Out] extends Logging
499503
this.dataQuanta().writeTextFileJava(url, formatterUdf, udfLoadProfileEstimator)
500504
}
501505

506+
/**
507+
* Feed the built [[DataQuanta]] into a [[org.apache.wayang.basic.operators.IcebergTableSink]]. This triggers
508+
* execution of the constructed [[WayangPlan]].
509+
*
510+
* @param catalog Iceberg Catalog
511+
* @param schema Iceberg Schema of the table to create
512+
* @param tableIdentifier Iceberg Table Identifier of the table to create
513+
* @param outputFileFormat File format of the output data files
514+
* @return the collected data quanta
515+
*/
516+
517+
def writeIcebergTable(catalog: Catalog,
518+
schema: Schema,
519+
tableIdentifier: TableIdentifier,
520+
outputFileFormat: FileFormat,
521+
jobName: String): Unit = {
522+
this.javaPlanBuilder.withJobName(jobName)
523+
this.dataQuanta().writeIcebergTableJava(catalog, schema, tableIdentifier, outputFileFormat)
524+
525+
}
526+
502527
/**
503528
* Feed the built [[DataQuanta]] into a [[org.apache.wayang.basic.operators.KafkaTopicSink]]. This triggers
504529
* execution of the constructed [[WayangPlan]].

‎wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/JavaPlanBuilder.scala‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,15 @@ import java.util.{Collection => JavaCollection}
2525
import org.apache.commons.lang3.Validate
2626
import org.apache.wayang.api.util.DataQuantaBuilderCache
2727
import org.apache.wayang.basic.data.Record
28-
import org.apache.wayang.basic.operators.{AmazonS3Source, AzureBlobStorageSource, GoogleCloudStorageSource, KafkaTopicSource, ParquetSource, TableSource, TextFileSource}
28+
import org.apache.wayang.basic.operators.{AmazonS3Source, AzureBlobStorageSource, GoogleCloudStorageSource, KafkaTopicSource, ParquetSource, TableSource, TextFileSource, ApacheIcebergSource}
2929
import org.apache.wayang.commons.util.profiledb.model.Experiment
3030
import org.apache.wayang.core.api.WayangContext
3131
import org.apache.wayang.core.plan.wayangplan._
3232
import org.apache.wayang.core.types.DataSetType
3333

3434
import scala.reflect.ClassTag
35+
import org.apache.iceberg.catalog.{Catalog, TableIdentifier}
36+
import org.apache.iceberg.expressions.Expression
3537

3638
/**
3739
* Utility to build and execute [[WayangPlan]]s.
@@ -79,6 +81,22 @@ class JavaPlanBuilder(wayangCtx: WayangContext, jobName: String) {
7981
preferDataset: Boolean = false): UnarySourceDataQuantaBuilder[UnarySourceDataQuantaBuilder[_, Record], Record] =
8082
createSourceBuilder(ParquetSource.create(url, projection).preferDatasetOutput(preferDataset))(ClassTag(classOf[Record]))
8183

84+
/**
85+
* Read an Apache Iceberg table and provide it as a dataset of [[Record]]s.
86+
*
87+
* @param catalog the Iceberg catalog containing the table
88+
* @param tableIdentifier the identifier of the Iceberg table to read
89+
* @param filterExpressions optional array of filter expressions to apply during the read
90+
* @param projectionColumns optional array of column names to project (select specific columns)
91+
* @return [[DataQuantaBuilder]] for the Iceberg table
92+
*/
93+
def readApacheIcebergTable(
94+
catalog: Catalog,
95+
tableIdentifier: TableIdentifier,
96+
filterExpressions: Array[Expression] = null,
97+
projectionColumns: Array[String] = null): UnarySourceDataQuantaBuilder[UnarySourceDataQuantaBuilder[_, Record], Record] =
98+
createSourceBuilder(ApacheIcebergSource.create(catalog, tableIdentifier, filterExpressions, projectionColumns))(ClassTag(classOf[Record]))
99+
82100
/**
83101
* Read a text file from a Google Cloud Storage bucket and provide it as a dataset of [[String]]s, one per line.
84102
*

‎wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/PlanBuilder.scala‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ package org.apache.wayang.api
2424
import org.apache.commons.lang3.Validate
2525
import org.apache.wayang.api
2626
import org.apache.wayang.basic.data.Record
27-
import org.apache.wayang.basic.operators.{AmazonS3Source, AzureBlobStorageSource, CollectionSource, GoogleCloudStorageSource, ObjectFileSource, ParquetSource, TableSource, TextFileSource}
27+
import org.apache.wayang.basic.operators.{AmazonS3Source, AzureBlobStorageSource, CollectionSource, GoogleCloudStorageSource, ObjectFileSource, ParquetSource, TableSource, TextFileSource, ApacheIcebergSource}
2828
import org.apache.wayang.commons.util.profiledb.model.Experiment
2929
import org.apache.wayang.core.api.WayangContext
3030
import org.apache.wayang.core.plan.wayangplan._
@@ -34,6 +34,8 @@ import scala.collection.JavaConversions
3434
import scala.collection.mutable.ListBuffer
3535
import scala.language.implicitConversions
3636
import scala.reflect._
37+
import org.apache.iceberg.catalog.{Catalog, TableIdentifier}
38+
import org.apache.iceberg.expressions.Expression
3739

3840
/**
3941
* Utility to build [[WayangPlan]]s.
@@ -144,6 +146,21 @@ class PlanBuilder(private[api] val wayangContext: WayangContext, private var job
144146
preferDataset: Boolean = false): DataQuanta[Record] =
145147
load(ParquetSource.create(url, projection).preferDatasetOutput(preferDataset))
146148

149+
/**
150+
* Read an Apache Iceberg table and provide it as a dataset of [[Record]]s.
151+
*
152+
* @param catalog the Iceberg catalog containing the table
153+
* @param tableIdentifier the identifier of the Iceberg table to read
154+
* @param filterExpressions optional array of filter expressions to apply during the read
155+
* @param projectionColumns optional array of column names to project (select specific columns)
156+
* @return [[DataQuanta]] of [[Record]] for the Iceberg table
157+
*/
158+
def readApacheIcebergTable(
159+
catalog: Catalog,
160+
tableIdentifier: TableIdentifier,
161+
filterExpressions: Array[Expression] = null,
162+
projectionColumns: Array[String] = null): DataQuanta[Record] = load(ApacheIcebergSource.create(catalog, tableIdentifier, filterExpressions, projectionColumns))
163+
147164
/**
148165
* Read a text file from a Google Cloud Storage bucket and provide it as a dataset of [[String]]s, one per line.
149166
*

‎wayang-commons/wayang-basic/pom.xml‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,11 @@
3333
This modules represents the base Wayang package with the default operators and functions.
3434
</description>
3535

36+
<properties>
37+
<iceberg.version>1.6.0</iceberg.version>
38+
<hadoop.version>3.3.6</hadoop.version>
39+
</properties>
40+
3641
<dependencyManagement>
3742
<dependencies>
3843
<!-- Google Cloud Libraries BOM -->
@@ -131,6 +136,45 @@
131136
<groupId>com.azure</groupId>
132137
<artifactId>azure-identity</artifactId>
133138
</dependency>
139+
140+
141+
<!-- Apache Iceberg properties-->
142+
143+
<dependency>
144+
<groupId>org.apache.iceberg</groupId>
145+
<artifactId>iceberg-core</artifactId>
146+
<version>${iceberg.version}</version>
147+
</dependency>
148+
<dependency>
149+
<groupId>org.apache.iceberg</groupId>
150+
<artifactId>iceberg-api</artifactId>
151+
<version>${iceberg.version}</version>
152+
</dependency>
153+
<dependency>
154+
<groupId>org.apache.iceberg</groupId>
155+
<artifactId>iceberg-parquet</artifactId>
156+
<version>${iceberg.version}</version>
157+
</dependency>
158+
<dependency>
159+
<groupId>org.apache.iceberg</groupId>
160+
<artifactId>iceberg-data</artifactId>
161+
<version>${iceberg.version}</version>
162+
</dependency>
163+
<dependency>
164+
<groupId>org.apache.hadoop</groupId>
165+
<artifactId>hadoop-common</artifactId>
166+
<version>${hadoop.version}</version>
167+
</dependency>
168+
<dependency>
169+
<groupId>org.apache.hadoop</groupId>
170+
<artifactId>hadoop-client</artifactId>
171+
<version>${hadoop.version}</version>
172+
</dependency>
173+
<dependency>
174+
<groupId>org.slf4j</groupId>
175+
<artifactId>slf4j-simple</artifactId>
176+
<version>2.0.16</version>
177+
</dependency>
134178

135179

136180
</dependencies>

‎wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/ApacheIcebergSink.java‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.wayang.basic.operators;
19+
20+
import org.apache.wayang.core.plan.wayangplan.UnarySink;
21+
import org.apache.wayang.core.types.DataSetType;
22+
23+
import org.apache.iceberg.Schema;
24+
import org.apache.iceberg.catalog.Catalog;
25+
import org.apache.iceberg.catalog.TableIdentifier;
26+
import org.apache.iceberg.FileFormat;
27+
import org.apache.wayang.basic.data.Record;
28+
29+
/**
30+
* This {@link UnarySink} writes all incoming data quanta to an iceberg table.
31+
* Either if the table does not exists it will create new, otherwise append.
32+
*
33+
* @param <T> Data Type if the incoming Data Quanta
34+
*/
35+
public class ApacheIcebergSink extends UnarySink<org.apache.wayang.basic.data.Record> {
36+
37+
protected final Catalog catalog;
38+
protected final Schema schema;
39+
protected final TableIdentifier tableIdentifier;
40+
protected final FileFormat outputFileFormat;
41+
42+
/**
43+
*
44+
* @param catalog Iceberg catalog used to resolve the target table; must
45+
* not be {@code null}
46+
* @param schema Iceberg write schema; must be compatible with the
47+
* target table
48+
* @param tableIdentifier fully qualified identifier of the target table
49+
*
50+
* @param outputFileFormat {@link FileFormat} the format of the output data files
51+
*/
52+
public ApacheIcebergSink(Catalog catalog, Schema schema, TableIdentifier tableIdentifier, FileFormat outputFileFormat) {
53+
super(DataSetType.createDefault(Record.class));
54+
this.catalog = catalog;
55+
this.schema = schema;
56+
this.tableIdentifier = tableIdentifier;
57+
this.outputFileFormat = outputFileFormat;
58+
}
59+
60+
/**
61+
* Creates a copied instance.
62+
*
63+
* @param that should be copied
64+
*/
65+
public ApacheIcebergSink(ApacheIcebergSink that) {
66+
super(that);
67+
this.catalog = that.catalog;
68+
this.schema = that.schema;
69+
this.tableIdentifier = that.tableIdentifier;
70+
this.outputFileFormat = that.outputFileFormat;
71+
}
72+
73+
public FileFormat getOutputFileFormat() {
74+
return this.outputFileFormat;
75+
}
76+
}

0 commit comments

Comments
 (0)