UPD: Migrate to parquet files

[vkhodygo]

I didn't want to invest too much of my time to get this working so I had to ask Claude to rework that code snippet a few times iteratively improving the prompt:

Dependencies

<dependencies>
    <dependency>
        <groupId>org.apache.parquet</groupId>
        <artifactId>parquet-hadoop</artifactId>
        <version>1.13.1</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.30</version>
    </dependency>
    <dependency>
        <groupId>com.fasterxml.jackson.dataformat</groupId>
        <artifactId>jackson-dataformat-yaml</artifactId>
        <version>2.15.3</version>
    </dependency>
</dependencies>

Schema

version: "1.0"
table: "population_data"
fields:
  - name: "year"
    type: "uint16"
    nullable: false
    description: "Year of the population record"
  
  - name: "country"
    type: "string"
    nullable: false
    description: "Country identifier"
  
  - name: "demographics"
    type: "struct"
    nullable: true
    fields:
      - name: "ageGroup"
        type: "uint8"
        nullable: false
      - name: "gender"
        type: "string"
        nullable: false
      - name: "count"
        type: "uint32"
        nullable: false

  - name: "metadata"
    type: "struct"
    nullable: true
    fields:
      - name: "source"
        type: "string"
        nullable: true
      - name: "lastUpdated"
        type: "timestamp"
        nullable: true

Schema configuration

// schema configuration classes
package com.example.schema;

import lombok.Getter;
import lombok.Setter;
import java.util.List;

@Getter
@Setter
public class SchemaConfig {
    private String version;
    private String table;
    private List<FieldConfig> fields;
}

@Getter
@Setter
public class FieldConfig {
    private String name;
    private String type;
    private boolean nullable;
    private String description;
    private List<FieldConfig> fields; // For nested structures
}

// data type mapping utility
package com.example.util;

import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
import org.apache.parquet.schema.Type;
import org.apache.parquet.schema.Types;

public class ParquetTypeMapper {
    public static Type mapField(FieldConfig field) {
        var builder = Types.required(mapDataType(field.getType()));
        
        if (field.isNullable()) {
            builder = Types.optional(mapDataType(field.getType()));
        }
        
        return builder.named(field.getName());
    }
    
    private static PrimitiveTypeName mapDataType(String type) {
        return switch (type.toLowerCase()) {
            case "uint8", "uint16" -> PrimitiveTypeName.INT32;
            case "uint32" -> PrimitiveTypeName.INT64;
            case "string" -> PrimitiveTypeName.BINARY;
            case "timestamp" -> PrimitiveTypeName.INT96;
            case "double" -> PrimitiveTypeName.DOUBLE;
            case "boolean" -> PrimitiveTypeName.BOOLEAN;
            default -> throw new IllegalArgumentException("Unsupported type: " + type);
        };
    }
}

// Main Parquet writer implementation
package com.example;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.ParquetFileWriter;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.MessageTypeParser;
import java.io.File;
import java.io.IOException;
import java.util.List;

@Slf4j
public class PopulationDataParquetWriter {
    private final SchemaConfig schemaConfig;
    private final String outputPath;
    
    public PopulationDataParquetWriter(String schemaPath, String outputPath) throws IOException {
        var mapper = new ObjectMapper(new YAMLFactory());
        this.schemaConfig = mapper.readValue(new File(schemaPath), SchemaConfig.class);
        this.outputPath = outputPath;
    }
    
    public void writePopulationData(String country, String inputFileName, int minStartYear, int maxStartYear) throws IOException {
        var schema = buildParquetSchema();
        
        try (var writer = createParquetWriter(schema)) {
            // Read CSV data similar to your original DataParser
            try (var reader = new CSVReader(new FileReader(inputFileName))) {
                String[] line;
                while ((line = reader.readNext()) != null) {
                    var record = convertToParquetRecord(line, schema);
                    writer.write(record);
                }
            }
        }
    }
    
    private MessageType buildParquetSchema() {
        var schemaBuilder = new StringBuilder();
        schemaBuilder.append("message ").append(schemaConfig.getTable()).append(" {\n");
        
        for (var field : schemaConfig.getFields()) {
            appendField(schemaBuilder, field, 1);
        }
        
        schemaBuilder.append("}");
        return MessageTypeParser.parseMessageType(schemaBuilder.toString());
    }
    
    private void appendField(StringBuilder builder, FieldConfig field, int depth) {
        var indent = "  ".repeat(depth);
        
        if ("struct".equals(field.getType())) {
            builder.append(indent)
                   .append("optional group ")
                   .append(field.getName())
                   .append(" {\n");
            
            for (var nestedField : field.getFields()) {
                appendField(builder, nestedField, depth + 1);
            }
            
            builder.append(indent).append("}\n");
        } else {
            builder.append(indent)
                   .append(field.isNullable() ? "optional " : "required ")
                   .append(ParquetTypeMapper.mapField(field).toString())
                   .append("\n");
        }
    }
    
    private ParquetWriter<?> createParquetWriter(MessageType schema) throws IOException {
        return ParquetWriter.builder(new Path(outputPath))
            .withSchema(schema)
            .withCompressionCodec(CompressionCodecName.SNAPPY)
            .withRowGroupSize(ParquetWriter.DEFAULT_BLOCK_SIZE)
            .withPageSize(ParquetWriter.DEFAULT_PAGE_SIZE)
            .withDictionaryPageSize(ParquetWriter.DEFAULT_PAGE_SIZE)
            .withDictionaryEncoding(true)
            .withValidation(true)
            .withWriterVersion(ParquetWriter.DEFAULT_WRITER_VERSION)
            .withConf(new Configuration())
            .build();
    }
}

// Updated main code replacing the original H2 database generation
public class DataGenerator {
    public static void generatePopulationData(Country country) {
        try {
            var schemaPath = "schema.yml";
            var outputPath = "./output/population_" + country.toString() + ".parquet";
            
            var writer = new PopulationDataParquetWriter(schemaPath, outputPath);
            writer.writePopulationData(
                country,
                Parameters.getPopulationInitialisationInputFileName(),
                Parameters.getMinStartYear(),
                Parameters.getMaxStartYear()
            );
            
        } catch (Exception e) {
            log.error("Failed to generate population data", e);
            throw new RuntimeException("Failed to generate population data", e);
        }
    }
}

There are other implementations, I can add them for comparison as well. I didn't validate any of that, but it seems to be legit. I

Data types can and will cause some problems. For example, to store UInt32 you can only employ Int64; this barely affects the actual file size as most of the data are redundant zeroes anyway, and they are compressed very well. However, reading into memory will blow up the dataset size.

One way to deal with this in this particular case is to use Int32, most likely all data values are well below the MAX_INT32.

One more issue here is that the number of data types in the Java implementation is very limited. For the initialisation stage this is less relevant, but this will affect the output size. I say, you pass the output directly to an R script which can further decrease the dataset size by employing more compact data types like Float16 and such.

[vkhodygo]

Note that nullable data columns take more space since they all have hidden indicator columns. If there is no nulls in a particular column it should be marked as such.

[vkhodygo]

An improved version with Hive and S3 support (you'll thank me later for the latter). Although I just noticed it's still using H2.

# Schema configuration for population data
schemaName: population_data
targetStorage:
  type: local  # Can be 'local', 'hive', or 's3'
  s3Config:
    bucket: population-data-bucket
    region: us-west-2
    accessKey: ${S3_ACCESS_KEY}
    secretKey: ${S3_SECRET_KEY}
  hiveConfig:
    warehouse: /user/hive/warehouse/population
    database: demographics

tables:
  - name: population_initial
    description: "Initial population data by year and country"
    partitionKeys:
      - country
      - year
    columns:
      - name: id
        type: long
        nullable: false
      - name: country
        type: string
        nullable: false
        isPartitionKey: true
      - name: year
        type: int
        nullable: false
        isPartitionKey: true
      - name: demographics:
        type: struct
        nullable: true
        fields:
          - name: age
            type: int
            nullable: false
          - name: gender
            type: string
            nullable: false
          - name: region
            type: string
            nullable: true
      - name: metrics
        type: struct
        nullable: true
        fields:
          - name: population_count
            type: long
            nullable: false
          - name: growth_rate
            type: double
            nullable: true

import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.avro.AvroParquetWriter;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.yaml.snakeyaml.Yaml;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import org.apache.hadoop.fs.FileSystem;
import java.util.*;
import java.io.*;

public class ParquetGenerator {
    private static final int BLOCK_SIZE = 256 * 1024 * 1024;
    private static final int PAGE_SIZE = 64 * 1024;

    private final SchemaConfig schemaConfig;
    private final StorageConfig storageConfig;

    public ParquetGenerator(String yamlConfigPath) throws IOException {
        this.schemaConfig = loadConfiguration(yamlConfigPath);
        this.storageConfig = schemaConfig.getTargetStorage();
    }

    // Configuration classes
    @Data
    public static class SchemaConfig {
        private String schemaName;
        private StorageConfig targetStorage;
        private List<TableConfig> tables;
    }

    @Data
    public static class StorageConfig {
        private String type;
        private S3Config s3Config;
        private HiveConfig hiveConfig;
    }

    @Data
    public static class S3Config {
        private String bucket;
        private String region;
        private String accessKey;
        private String secretKey;
    }

    @Data
    public static class HiveConfig {
        private String warehouse;
        private String database;
    }

    @Data
    public static class TableConfig {
        private String name;
        private String description;
        private List<String> partitionKeys;
        private List<ColumnConfig> columns;
    }

    @Data
    public static class ColumnConfig {
        private String name;
        private String type;
        private boolean nullable;
        private boolean isPartitionKey;
        private List<ColumnConfig> fields;
    }

    // Main conversion method
    public void convertToParquet(Connection h2Conn, String country, int minYear, int maxYear) throws Exception {
        for (TableConfig tableConfig : schemaConfig.getTables()) {
            Schema avroSchema = convertToAvroSchema(tableConfig);
            
            for (int year = minYear; year <= maxYear; year++) {
                String outputPath = buildOutputPath(tableConfig.getName(), country, year);
                
                try (ParquetWriter<GenericRecord> writer = createParquetWriter(outputPath, avroSchema)) {
                    processData(h2Conn, writer, country, year, tableConfig);
                }
            }
        }
    }

    // Create Parquet writer based on storage configuration
    private ParquetWriter<GenericRecord> createParquetWriter(String outputPath, Schema avroSchema) throws IOException {
        Path path = new Path(outputPath);
        Configuration conf = new Configuration();

        switch (storageConfig.getType().toLowerCase()) {
            case "s3":
                setupS3Configuration(conf);
                break;
            case "hive":
                setupHiveConfiguration(conf);
                break;
            default:
                // Local filesystem configuration
                break;
        }

        return AvroParquetWriter.<GenericRecord>builder(path)
                .withSchema(avroSchema)
                .withConf(conf)
                .withCompressionCodec(CompressionCodecName.SNAPPY)
                .withBlockSize(BLOCK_SIZE)
                .withPageSize(PAGE_SIZE)
                .build();
    }

    // Setup S3 configuration
    private void setupS3Configuration(Configuration conf) {
        S3Config s3Config = storageConfig.getS3Config();
        conf.set("fs.s3a.access.key", s3Config.getAccessKey());
        conf.set("fs.s3a.secret.key", s3Config.getSecretKey());
        conf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");
    }

    // Setup Hive configuration
    private void setupHiveConfiguration(Configuration conf) {
        HiveConfig hiveConfig = storageConfig.getHiveConfig();
        conf.set("hive.metastore.warehouse.dir", hiveConfig.getWarehouse());
        conf.set("hive.metastore.uris", "thrift://localhost:9083");
    }

    // Convert YAML schema to Avro schema
    private Schema convertToAvroSchema(TableConfig tableConfig) {
        Schema.Parser parser = new Schema.Parser();
        JsonNodeFactory factory = JsonNodeFactory.instance;
        ObjectNode schemaNode = factory.objectNode();
        
        schemaNode.put("type", "record");
        schemaNode.put("name", tableConfig.getName());
        schemaNode.put("namespace", schemaConfig.getSchemaName());

        ArrayNode fieldsNode = factory.arrayNode();
        for (ColumnConfig column : tableConfig.getColumns()) {
            fieldsNode.add(convertColumnToAvroField(column));
        }
        schemaNode.set("fields", fieldsNode);

        return parser.parse(schemaNode.toString());
    }

    // Convert column configuration to Avro field
    private ObjectNode convertColumnToAvroField(ColumnConfig column) {
        ObjectNode fieldNode = JsonNodeFactory.instance.objectNode();
        fieldNode.put("name", column.getName());
        
        if (column.getType().equals("struct")) {
            ArrayNode subFields = JsonNodeFactory.instance.arrayNode();
            for (ColumnConfig subColumn : column.getFields()) {
                subFields.add(convertColumnToAvroField(subColumn));
            }
            
            ObjectNode structType = JsonNodeFactory.instance.objectNode();
            structType.put("type", "record");
            structType.put("name", column.getName() + "Record");
            structType.set("fields", subFields);
            
            if (column.isNullable()) {
                ArrayNode typeArray = JsonNodeFactory.instance.arrayNode();
                typeArray.add("null");
                typeArray.add(structType);
                fieldNode.set("type", typeArray);
            } else {
                fieldNode.set("type", structType);
            }
        } else {
            String avroType = convertToAvroType(column.getType());
            if (column.isNullable()) {
                ArrayNode typeArray = JsonNodeFactory.instance.arrayNode();
                typeArray.add("null");
                typeArray.add(avroType);
                fieldNode.set("type", typeArray);
            } else {
                fieldNode.put("type", avroType);
            }
        }
        
        return fieldNode;
    }

    // Process data from H2 and write to Parquet
    private void processData(Connection h2Conn, ParquetWriter<GenericRecord> writer, 
                           String country, int year, TableConfig tableConfig) throws Exception {
        String sql = buildSelectQuery(tableConfig, country, year);
        
        try (PreparedStatement stmt = h2Conn.prepareStatement(sql);
             ResultSet rs = stmt.executeQuery()) {
            
            while (rs.next()) {
                GenericRecord record = createRecordFromResultSet(rs, tableConfig);
                writer.write(record);
            }
        }
    }

    // Build output path based on storage configuration
    private String buildOutputPath(String tableName, String country, int year) {
        StringBuilder path = new StringBuilder();
        
        switch (storageConfig.getType().toLowerCase()) {
            case "s3":
                path.append("s3a://")
                    .append(storageConfig.getS3Config().getBucket())
                    .append("/");
                break;
            case "hive":
                path.append(storageConfig.getHiveConfig().getWarehouse())
                    .append("/")
                    .append(storageConfig.getHiveConfig().getDatabase())
                    .append("/");
                break;
            default:
                path.append("file:///data/");
        }
        
        path.append(tableName)
            .append("/country=").append(country)
            .append("/year=").append(year)
            .append("/data.parquet");
            
        return path.toString();
    }

    // Type conversion utility
    private String convertToAvroType(String yamlType) {
        switch (yamlType.toLowerCase()) {
            case "long":
                return "long";
            case "int":
                return "int";
            case "string":
                return "string";
            case "double":
                return "double";
            case "boolean":
                return "boolean";
            case "binary":
                return "bytes";
            default:
                throw new IllegalArgumentException("Unsupported type: " + yamlType);
        }
    }

    // Main usage example
    public static void main(String[] args) {
        try {
            // Initialize generator with YAML config
            ParquetGenerator generator = new ParquetGenerator("schema-config.yaml");
            
            // Setup H2 connection (using your existing code)
            Connection conn = null;
            try {
                Class.forName("org.h2.Driver");
                conn = DriverManager.getConnection("jdbc:h2:file:./input" + File.separator + "input", "sa", "");
                
                // Convert data to Parquet
                generator.convertToParquet(conn, "USA", 2020, 2025);
                
            } finally {
                if (conn != null) {
                    conn.close();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

[vkhodygo]

v3

# Schema configuration for population data
fields:
  - name: id
    type: string
    partitionKey: false
  
  - name: country
    type: string
    partitionKey: true
  
  - name: year
    type: integer
    partitionKey: true
  
  - name: age
    type: integer
    partitionKey: false
  
  - name: gender
    type: string
    partitionKey: false
  
  - name: demographics
    type: nested
    partitionKey: false
    nestedFields:
      - name: education_level
        type: string
      - name: income_bracket
        type: string
      - name: employment_status
        type: string
  
  - name: population_count
    type: integer
    partitionKey: false
  
  - name: growth_rate
    type: double
    partitionKey: false

outputConfig:
  useHive: true
  hiveMetastoreUri: "thrift://localhost:9083"
  hiveWarehouseDir: "/user/hive/warehouse"
  
  useS3: true
  s3Bucket: "population-data"
  s3Key: "raw-data"

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.avro.AvroParquetWriter;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.yaml.snakeyaml.Yaml;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import java.io.*;
import java.util.*;

public class ParquetDataConverter {
    private final SchemaConfig schemaConfig;
    private final OutputConfig outputConfig;
    
    public ParquetDataConverter(String schemaPath) throws IOException {
        this.schemaConfig = loadSchemaConfig(schemaPath);
        this.outputConfig = schemaConfig.getOutputConfig();
    }

    // Main method to replace the original H2 database creation
    public void convertCSVToParquet(String country, String inputFileName, int minStartYear, int maxStartYear) throws IOException {
        for (int year = minStartYear; year <= maxStartYear; year++) {
            String inputPath = String.format("input/%s_%d.csv", inputFileName, year);
            String outputPath = buildOutputPath(country, year);
            
            processCSVFile(inputPath, outputPath, year);
        }
    }

    private void processCSVFile(String inputPath, String outputPath, int year) throws IOException {
        try (Reader reader = new FileReader(inputPath);
             CSVParser csvParser = new CSVParser(reader, CSVFormat.DEFAULT.withHeader())) {
            
            // Create Avro schema from YAML configuration
            Schema avroSchema = createAvroSchema();
            
            // Setup writer configuration
            Configuration conf = new Configuration();
            if (outputConfig.isUseHive()) {
                setupHiveConfiguration(conf);
            }
            
            // Create Parquet writer
            try (ParquetWriter<GenericData.Record> writer = AvroParquetWriter
                    .<GenericData.Record>builder(new Path(outputPath))
                    .withSchema(avroSchema)
                    .withConf(conf)
                    .withCompressionCodec(CompressionCodecName.SNAPPY)
                    .withPageSize(4 * 1024 * 1024)
                    .withRowGroupSize(16 * 1024 * 1024)
                    .build()) {

                // Process each record
                for (CSVRecord record : csvParser) {
                    GenericData.Record avroRecord = new GenericData.Record(avroSchema);
                    
                    // Map CSV fields to Avro record based on schema
                    for (FieldConfig field : schemaConfig.getFields()) {
                        Object value = convertValue(record.get(field.getName()), field);
                        avroRecord.put(field.getName(), value);
                    }
                    
                    writer.write(avroRecord);
                }
            }
            
            // If S3 is configured, upload the file
            if (outputConfig.isUseS3()) {
                uploadToS3(outputPath);
            }
        }
    }

    private Schema createAvroSchema() {
        // Build Avro schema from YAML configuration
        StringBuilder schemaJson = new StringBuilder();
        schemaJson.append("{\"type\":\"record\",\"name\":\"PopulationRecord\",\"fields\":[");
        
        List<String> fields = new ArrayList<>();
        for (FieldConfig field : schemaConfig.getFields()) {
            fields.add(String.format("{\"name\":\"%s\",\"type\":%s}", 
                field.getName(), 
                getAvroType(field)));
        }
        
        schemaJson.append(String.join(",", fields));
        schemaJson.append("]}");
        
        return new Schema.Parser().parse(schemaJson.toString());
    }

    private String getAvroType(FieldConfig field) {
        switch (field.getType().toLowerCase()) {
            case "string":
                return "\"string\"";
            case "integer":
                return "\"int\"";
            case "double":
                return "\"double\"";
            case "boolean":
                return "\"boolean\"";
            case "nested":
                return createNestedType(field);
            default:
                throw new IllegalArgumentException("Unsupported type: " + field.getType());
        }
    }

    private String createNestedType(FieldConfig field) {
        StringBuilder nested = new StringBuilder();
        nested.append("{\"type\":\"record\",\"name\":\"")
              .append(field.getName())
              .append("\",\"fields\":[");
        
        List<String> nestedFields = new ArrayList<>();
        for (FieldConfig nestedField : field.getNestedFields()) {
            nestedFields.add(String.format("{\"name\":\"%s\",\"type\":%s}", 
                nestedField.getName(), 
                getAvroType(nestedField)));
        }
        
        nested.append(String.join(",", nestedFields));
        nested.append("]}");
        return nested.toString();
    }

    private Object convertValue(String value, FieldConfig field) {
        if (value == null || value.isEmpty()) {
            return null;
        }
        
        switch (field.getType().toLowerCase()) {
            case "string":
                return value;
            case "integer":
                return Integer.parseInt(value);
            case "double":
                return Double.parseDouble(value);
            case "boolean":
                return Boolean.parseBoolean(value);
            case "nested":
                return convertNestedValue(value, field);
            default:
                throw new IllegalArgumentException("Unsupported type: " + field.getType());
        }
    }

    private GenericData.Record convertNestedValue(String value, FieldConfig field) {
        // Implementation for nested value conversion
        // This would need to be customized based on your nested data format
        return null;
    }

    private void setupHiveConfiguration(Configuration conf) {
        // Set Hive-specific configuration
        conf.set("hive.metastore.uris", outputConfig.getHiveMetastoreUri());
        conf.set("hive.metastore.warehouse.dir", outputConfig.getHiveWarehouseDir());
    }

    private void uploadToS3(String filePath) {
        AmazonS3 s3Client = AmazonS3ClientBuilder.defaultClient();
        File file = new File(filePath);
        s3Client.putObject(
            outputConfig.getS3Bucket(),
            outputConfig.getS3Key() + "/" + file.getName(),
            file
        );
    }

    private String buildOutputPath(String country, int year) {
        StringBuilder path = new StringBuilder();
        
        if (outputConfig.isUseHive()) {
            path.append(outputConfig.getHiveWarehouseDir())
                .append("/");
        }
        
        path.append("population_data");
        
        // Add partition directories
        for (FieldConfig field : schemaConfig.getFields()) {
            if (field.isPartitionKey()) {
                path.append("/")
                    .append(field.getName())
                    .append("=")
                    .append(field.getName().equals("country") ? country : year);
            }
        }
        
        path.append(".parquet");
        return path.toString();
    }

    private SchemaConfig loadSchemaConfig(String schemaPath) throws IOException {
        try (InputStream input = new FileInputStream(schemaPath)) {
            Yaml yaml = new Yaml();
            return yaml.loadAs(input, SchemaConfig.class);
        }
    }
}

// Configuration classes
class SchemaConfig {
    private List<FieldConfig> fields;
    private OutputConfig outputConfig;
    
    // Getters and setters
    public List<FieldConfig> getFields() { return fields; }
    public void setFields(List<FieldConfig> fields) { this.fields = fields; }
    public OutputConfig getOutputConfig() { return outputConfig; }
    public void setOutputConfig(OutputConfig outputConfig) { this.outputConfig = outputConfig; }
}

class FieldConfig {
    private String name;
    private String type;
    private boolean partitionKey;
    private List<FieldConfig> nestedFields;
    
    // Getters and setters
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public String getType() { return type; }
    public void setType(String type) { this.type = type; }
    public boolean isPartitionKey() { return partitionKey; }
    public void setPartitionKey(boolean partitionKey) { this.partitionKey = partitionKey; }
    public List<FieldConfig> getNestedFields() { return nestedFields; }
    public void setNestedFields(List<FieldConfig> nestedFields) { this.nestedFields = nestedFields; }
}

class OutputConfig {
    private boolean useHive;
    private String hiveMetastoreUri;
    private String hiveWarehouseDir;
    private boolean useS3;
    private String s3Bucket;
    private String s3Key;
    
    // Getters and setters
    public boolean isUseHive() { return useHive; }
    public void setUseHive(boolean useHive) { this.useHive = useHive; }
    public String getHiveMetastoreUri() { return hiveMetastoreUri; }
    public void setHiveMetastoreUri(String uri) { this.hiveMetastoreUri = uri; }
    public String getHiveWarehouseDir() { return hiveWarehouseDir; }
    public void setHiveWarehouseDir(String dir) { this.hiveWarehouseDir = dir; }
    public boolean isUseS3() { return useS3; }
    public void setUseS3(boolean useS3) { this.useS3 = useS3; }
    public String getS3Bucket() { return s3Bucket; }
    public void setS3Bucket(String bucket) { this.s3Bucket = bucket; }
    public String getS3Key() { return s3Key; }
    public void setS3Key(String key) { this.s3Key = key; }
}

// Required dependencies in pom.xml:
/*
<dependencies>
    <dependency>
        <groupId>org.apache.parquet</groupId>
        <artifactId>parquet-avro</artifactId>
        <version>1.13.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.3.6</version>
    </dependency>
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>1.11.3</version>
    </dependency>
    <dependency>
        <groupId>org.yaml</groupId>
        <artifactId>snakeyaml</artifactId>
        <version>2.2</version>
    </dependency>
    <dependency>
        <groupId>com.amazonaws</groupId>
        <artifactId>aws-java-sdk-s3</artifactId>
        <version>1.12.621</version>
    </dependency>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-csv</artifactId>
        <version>1.10.0</version>
    </dependency>
</dependencies>
*/

[vkhodygo]

I Just realised you still use ORM elsewhere in the code, so other parts have to be changed or updated as well.

See, for example, this https://www.cdata.com/kb/tech/parquet-python-sqlalchemy.rst https://www.cdata.com/kb/tech/parquet-jdbc-hibernate.rst