Support exports to Parquet format

.github/workflows/maven.yml

@@ -21,7 +21,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        java_version: ['11', '17', '21']
+        java_version: ['17', '21', '25']
         os: [ubuntu-latest]
     steps:
       - uses: actions/checkout@v4
@@ -44,27 +44,31 @@ jobs:
         uses: codecov/codecov-action@v4
         with:
           token: ${{ secrets.CODECOV_TOKEN }}
-          files: ./dbeam-core/target/site/jacoco/jacoco.xml
+          files: ./dbeam-core/target/site/jacoco/jacoco.xml,./dbeam-parquet/target/site/jacoco/jacoco.xml
           flags: unittests # optional
           name: codecov-umbrella # optional
           fail_ci_if_error: false # optional (default = false)
           verbose: true # optional (default = false)
-      - name: Install avro tools
-        if: matrix.java_version == '17'
+      - name: Install avro and parquet tools
+        if: matrix.java_version == '21'
         run: |
-          wget https://dlcdn.apache.org/avro/avro-1.11.3/java/avro-tools-1.11.3.jar
+          wget https://dlcdn.apache.org/avro/avro-1.11.5/java/avro-tools-1.11.5.jar
           mkdir -p /opt/avro
-          mv avro-tools-1.11.3.jar /opt/avro/
-          chmod +x /opt/avro/avro-tools-1.11.3.jar
-          echo "alias avro-tools='java -jar /opt/avro/avro-tools-1.11.3.jar'" > ~/.bashrc
+          mv avro-tools-1.11.5.jar /opt/avro/
+          chmod +x /opt/avro/avro-tools-1.11.5.jar
+          echo "alias avro-tools='java -jar /opt/avro/avro-tools-1.11.5.jar'" > ~/.bashrc
+          wget -q https://repo1.maven.org/maven2/org/apache/parquet/parquet-cli/1.14.4/parquet-cli-1.14.4-runtime.jar
+          mkdir -p /opt/parquet
+          mv parquet-cli-1.14.4-runtime.jar /opt/parquet/
+          echo "alias parquet-tools='java -jar /opt/parquet/parquet-cli-1.14.4-runtime.jar'" >> ~/.bashrc
       - name: End to end tests
         shell: bash
-        if: matrix.java_version == '17'
+        if: matrix.java_version == '21'
         run: ./e2e/e2e.sh
       - run: mkdir staging && cp /tmp/debeam_e2e.log ./staging
-        if: matrix.java_version == '17'
+        if: matrix.java_version == '21'
       - uses: actions/upload-artifact@v4
-        if: github.event_name == 'push' && (github.ref == 'refs/heads/master' || github.ref == 'refs/heads/main') && matrix.java_version == '17'
+        if: github.event_name == 'push' && (github.ref == 'refs/heads/master' || github.ref == 'refs/heads/main') && matrix.java_version == '21'
         with:
           name: Package
           path: staging
@@ -86,12 +90,12 @@ jobs:
         uses: actions/cache@v4
         with:
           path: ~/.m2
-          key: Linux-java11-m2-${{ hashFiles('**/pom.xml') }}
-          restore-keys: Linux-java11-m2
-      - name: Set up JDK 11
+          key: Linux-java21-m2-${{ hashFiles('**/pom.xml') }}
+          restore-keys: Linux-java21-m2
+      - name: Set up JDK 21
         uses: actions/setup-java@v4
         with:
-          java-version: '11'
+          java-version: '21'
           distribution: 'adopt'  # use Corretto once supported https://github.com/actions/setup-java/issues/68
       - name: Deploy
         env:

.github/workflows/release.yml

@@ -23,12 +23,12 @@ jobs:
         uses: actions/cache@v4
         with:
           path: ~/.m2
-          key: Linux-java11-m2-${{ hashFiles('**/pom.xml') }}
-          restore-keys: Linux-java11-m2
-      - name: Set up JDK 11
+          key: Linux-java17-m2-${{ hashFiles('**/pom.xml') }}
+          restore-keys: Linux-java17-m2
+      - name: Set up JDK 17
         uses: actions/setup-java@v4
         with:
-          java-version: '11'
+          java-version: '17'
           distribution: 'adopt'
           # setup-java generates a settings.xml pointing deployments to Sonatype
           # See https://github.com/actions/setup-java/blob/v3.11.0/docs/advanced-usage.md#publishing-using-apache-maven

CHANGELOG.md

@@ -0,0 +1,42 @@
+# Changelog
+
+## [Unreleased]
+
+### Added
+
+- **dbeam-parquet module**: New module for exporting SQL databases directly to Parquet files,
+  complementing the existing Avro export in dbeam-core.
+  - `JdbcParquetJob`: Main entry point for Parquet exports, reuses all shared infrastructure
+    from dbeam-core (JDBC connectors, query building, partitioning, parallel queries, Beam runners).
+  - `JdbcParquetSchema`: Converts JDBC ResultSetMetaData to Parquet MessageType schema with
+    full SQL type mapping including logical types (TIMESTAMP_MILLIS, UUID).
+  - `JdbcParquetWriteSupport`: Writes ResultSet rows directly to Parquet RecordConsumer
+    without intermediate objects.
+  - `ChannelOutputFile`: OutputFile implementation backed by WritableByteChannel, bypassing
+    Hadoop for file I/O.
+  - Parquet compression codecs: snappy, gzip, zstd, lz4, uncompressed via `--parquetCodec`
+    option (also auto-maps from `--avroCodec`).
+  - SQL ARRAY columns use Parquet 3-level LIST type with typed elements (INT32, INT64, FLOAT,
+    DOUBLE, BOOLEAN, STRING) inferred from column type name.
+  - Parquet file footer includes `parquet.avro.schema` key with the Avro schema JSON, generated
+    via JdbcAvroSchema for full compatibility with Spark, Hive, and BigQuery.
+  - `_AVRO_SCHEMA.avsc` file saved alongside `_PARQUET_SCHEMA.json` in output directory.
+  - `--parquetSchemaFilePath`: Input schema override (Parquet MessageType text format).
+  - `--rowGroupSize` (default 128MB) and `--pageSize` (default 1MB) options.
+  - `BeamJdbcParquetSchema`: Exposes schema creation timing as Beam metric.
+  - `PsqlParquetJob`: PostgreSQL-specific job with replication lag check.
+  - `BenchJdbcParquetJob`: Benchmarking job for multi-execution performance testing.
+  - End-to-end test support in `e2e/e2e.sh` with `parquet-tools` validation.
+  - Comprehensive test suite (90+ tests) covering schema conversion, record roundtrips,
+    typed arrays, null handling, mock-based PostgreSQL type tests, and footer metadata.
+
+### Changed
+
+- Parent POM: Shade plugin config parameterized via `${dbeam.mainClass}` property,
+  eliminating duplication between dbeam-core and dbeam-parquet.
+- `PsqlReplicationCheck.validateOptions()` made public for cross-module access.
+- dbeam-core: Added test-jar packaging for test helper sharing.
+- CI: Codecov upload includes dbeam-parquet coverage. Parquet-tools installed for e2e validation.
+- e2e: PostgreSQL upgraded from 16 to 18. Parquet test suite runs alongside Avro suite.
+- Documentation: README updated with Parquet usage examples, library dependencies, and feature list.
+  New `docs/parquet-type-conversion.md` with full type mapping table.

README.md

@@ -17,28 +17,38 @@ This tool is runnable locally, or on any other backend supported by Apache Beam,
 ## Overview
 
 DBeam is a tool that reads all the data from single SQL database table,
-converts the data into [Avro](https://avro.apache.org/) and stores it into
+converts the data into [Avro](https://avro.apache.org/) or [Parquet](https://parquet.apache.org/) and stores it into
 appointed location, usually in GCS.
 It runs as a single threaded [Apache Beam](https://beam.apache.org/) pipeline.
 
 DBeam requires the database credentials, the database table name to read, and the output location
 to store the extracted data into. DBeam first makes a single select into the target table with
 limit one to infer the table schema. After the schema is created the job will be launched which
-simply streams the table contents via JDBC into target location as Avro.
+simply streams the table contents via JDBC into target location as Avro or Parquet.
 
-[Generated Avro Schema Type Conversion Details](docs/type-conversion.md)
+Type conversion details: [Avro](docs/type-conversion.md) | [Parquet](docs/parquet-type-conversion.md)
 
 
 ## dbeam-core package features
 
 - Supports both PostgreSQL, MySQL, MariaDB, and H2 JDBC connectors
 - Supports [Google CloudSQL](https://cloud.google.com/sql/) managed databases
-- Currently outputs only to Avro format
+- Outputs to Avro format
 - Reads database from an external password file (`--passwordFile`) or an external [KMS](https://cloud.google.com/kms/) encrypted password file (`--passwordFileKmsEncrypted`)
 - Can filter only records of the current day with the `--partitionColumn` parameter
 - Check and fail on too old partition dates. Snapshot dumps are not filtered by a given date/partition, when running for a too old partition, the job fails to avoid new data in old partitions. (can be disabled with `--skipPartitionCheck`)
 - Implemented as [Apache Beam SDK](https://beam.apache.org/) pipeline, supporting any of its [runners](https://beam.apache.org/documentation/runners/capability-matrix/) (tested with `DirectRunner` and `DataflowRunner`)
 
+## dbeam-parquet package features
+
+- Outputs to Parquet format (columnar, optimized for analytical queries)
+- Reuses all dbeam-core infrastructure (JDBC connectors, CloudSQL, partitioning, parallel queries)
+- Supports Parquet compression codecs: snappy, gzip, zstd, lz4, uncompressed
+- No full Hadoop dependency — uses a minimal `hadoop-common` for the Parquet API surface
+- Separate `--parquetCodec` option or automatic mapping from `--avroCodec`
+- Optional input schema file via `--parquetSchemaFilePath` (Parquet MessageType text format)
+- PostgreSQL replication check via `PsqlParquetJob`
+
 ### DBeam export parameters
 
 ```
@@ -171,15 +181,19 @@ Building and testing can be achieved with `mvn`:
 mvn verify
 ```
 
-In order to create a jar with all dependencies under `./dbeam-core/target/dbeam-core-shaded.jar` run the following:
+In order to create jars with all dependencies run the following:
 
 ```sh
 mvn clean package -Ppack
 ```
 
+This produces:
+- `./dbeam-core/target/dbeam-core-shaded.jar` — Avro export
+- `./dbeam-parquet/target/dbeam-parquet-shaded.jar` — Parquet export
+
 ## Usage examples
 
-Using Java from the command line:
+### Avro export
 
 ```sh
 java -cp ./dbeam-core/target/dbeam-core-shaded.jar \
@@ -191,7 +205,20 @@ java -cp ./dbeam-core/target/dbeam-core-shaded.jar \
   --table=my_table
 ```
 
-For CloudSQL:
+### Parquet export
+
+```sh
+java -cp ./dbeam-parquet/target/dbeam-parquet-shaded.jar \
+  com.spotify.dbeam.parquet.JdbcParquetJob \
+  --output=gs://my-testing-bucket-name/ \
+  --username=my_database_username \
+  --password=secret \
+  --connectionUrl=jdbc:postgresql://some.database.uri.example.org:5432/my_database \
+  --table=my_table \
+  --parquetCodec=snappy
+```
+
+For CloudSQL (Avro):
 
 ```sh
 java -cp ./dbeam-core/target/dbeam-core-shaded.jar \
@@ -256,19 +283,28 @@ When using Google Cloud, [IAM authentication](https://github.com/GoogleCloudPlat
 To include DBeam library in a mvn project add the following dependency in `pom.xml`:
 
 ```xml
+<!-- For Avro export -->
 <dependency>
   <groupId>com.spotify</groupId>
   <artifactId>dbeam-core</artifactId>
   <version>${dbeam.version}</version>
 </dependency>
+
+<!-- For Parquet export -->
+<dependency>
+  <groupId>com.spotify</groupId>
+  <artifactId>dbeam-parquet</artifactId>
+  <version>${dbeam.version}</version>
+</dependency>
 ```
 
 
 To include DBeam library in a SBT project add the following dependency in `build.sbt`:
 
 ```sbt
   libraryDependencies ++= Seq(
-   "com.spotify" % "dbeam-core" % dbeamVersion
+   "com.spotify" % "dbeam-core" % dbeamVersion,     // Avro
+   "com.spotify" % "dbeam-parquet" % dbeamVersion    // Parquet
   )
 ```
 

dbeam-bom/pom.xml

@@ -21,6 +21,11 @@
         <artifactId>dbeam-core</artifactId>
         <version>${project.version}</version>
       </dependency>
+      <dependency>
+        <groupId>com.spotify</groupId>
+        <artifactId>dbeam-parquet</artifactId>
+        <version>${project.version}</version>
+      </dependency>
     </dependencies>
   </dependencyManagement>
 
@@ -47,7 +52,7 @@
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>build-helper-maven-plugin</artifactId>
-        <version>3.6.0</version>
+        <version>3.6.1</version>
         <executions>
           <execution>
             <id>attach-effective-pom</id>

dbeam-core/pom.xml

@@ -13,6 +13,10 @@
   <name>DBeam Core</name>
   <description>Top level DBeam core implementation</description>
 
+  <properties>
+    <dbeam.mainClass>com.spotify.dbeam.jobs.JdbcAvroJob</dbeam.mainClass>
+  </properties>
+
   <dependencies>
     <!-- SDK -->
     <dependency>
@@ -93,8 +97,8 @@
     <!-- Test Deps -->
 
     <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter</artifactId>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -124,4 +128,20 @@
     </dependency>
   </dependencies>
 
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-jar-plugin</artifactId>
+        <executions>
+          <execution>
+            <goals>
+              <goal>test-jar</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+
 </project>

dbeam-core/src/main/java/com/spotify/dbeam/args/JdbcAvroArgs.java

@@ -103,8 +103,8 @@ public static JdbcAvroArgs create(
   }
 
   public static JdbcAvroArgs create(final JdbcConnectionArgs jdbcConnectionArgs) {
-    return create(jdbcConnectionArgs, 10000, "deflate6", Collections.emptyList(),
-        "typed_first_row", false);
+    return create(
+        jdbcConnectionArgs, 10000, "deflate6", Collections.emptyList(), "typed_first_row", false);
   }
 
   public interface StatementPreparator extends Serializable {

dbeam-core/src/main/java/com/spotify/dbeam/avro/JdbcAvroIO.java

@@ -25,7 +25,6 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.io.CountingOutputStream;
 import com.spotify.dbeam.args.JdbcAvroArgs;
-import com.spotify.dbeam.args.JdbcExportArgs;
 import java.nio.channels.Channels;
 import java.nio.channels.WritableByteChannel;
 import java.sql.Connection;
@@ -137,7 +136,7 @@ private static class JdbcAvroWriter extends FileBasedSink.Writer<Void, String> {
     private final JdbcAvroArgs jdbcAvroArgs;
     private DataFileWriter<GenericRecord> dataFileWriter;
     private Connection connection;
-    private JdbcAvroMetering metering;
+    private JdbcMetering metering;
     private CountingOutputStream countingOutputStream;
 
     JdbcAvroWriter(
@@ -147,7 +146,7 @@ private static class JdbcAvroWriter extends FileBasedSink.Writer<Void, String> {
       super(writeOperation, MimeTypes.BINARY);
       this.dynamicDestinations = dynamicDestinations;
       this.jdbcAvroArgs = jdbcAvroArgs;
-      this.metering = JdbcAvroMetering.create();
+      this.metering = JdbcMetering.create("jdbcavroio");
     }
 
     public Void getDestination() {
@@ -181,8 +180,7 @@ private ResultSet executeQuery(final String query) throws Exception {
         jdbcAvroArgs.statementPreparator().setParameters(statement);
       }
 
-      if (jdbcAvroArgs.preCommand() != null
-          && !jdbcAvroArgs.preCommand().isEmpty()) {
+      if (jdbcAvroArgs.preCommand() != null && !jdbcAvroArgs.preCommand().isEmpty()) {
         final Statement stmt = connection.createStatement();
         for (String command : jdbcAvroArgs.preCommand()) {
           stmt.execute(command);
@@ -205,8 +203,9 @@ public void write(final String query) throws Exception {
       LOGGER.info("jdbcavroio : Starting write...");
       try (ResultSet resultSet = executeQuery(query)) {
         metering.startWriteMeter();
-        final JdbcAvroRecordConverter converter = JdbcAvroRecordConverter.create(resultSet,
-            this.jdbcAvroArgs.arrayMode(), this.jdbcAvroArgs.nullableArrayItems());
+        final JdbcAvroRecordConverter converter =
+            JdbcAvroRecordConverter.create(
+                resultSet, this.jdbcAvroArgs.arrayMode(), this.jdbcAvroArgs.nullableArrayItems());
         while (resultSet.next()) {
           dataFileWriter.appendEncoded(converter.convertResultSetIntoAvroBytes());
           this.metering.incrementRecordCount();