You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2023/05/31 07:47:37 UTC
[camel] branch main updated: CAMEL-13573 : create parquet-avro dataformat (#10218)
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new be3793a67b8 CAMEL-13573 : create parquet-avro dataformat (#10218)
be3793a67b8 is described below
commit be3793a67b83c5b7ffc3390e504565efacba335a
Author: önder sezgin <on...@gmail.com>
AuthorDate: Wed May 31 08:47:30 2023 +0100
CAMEL-13573 : create parquet-avro dataformat (#10218)
* CAMEL-13573 : create parquet-avro dataformat
* address review comments
* fix cs
* fix cs
* fix cs
* fix cs
* fix cs
---------
Co-authored-by: Andrea Cosentino <an...@gmail.com>
---
camel-dependencies/pom.xml | 2 +
catalog/camel-allcomponents/pom.xml | 5 +
.../org/apache/camel/catalog/models.properties | 1 +
.../org/apache/camel/catalog/models/marshal.json | 2 +-
.../apache/camel/catalog/models/parquetAvro.json | 18 +++
.../org/apache/camel/catalog/models/unmarshal.json | 2 +-
.../apache/camel/catalog/schemas/camel-spring.xsd | 38 ++++++-
components/camel-parquet-avro/pom.xml | 89 +++++++++++++++
.../org/apache/camel/dataformat.properties | 7 ++
.../org/apache/camel/dataformat/parquetAvro | 2 +
.../src/main/doc/parquetAvro-dataformat.adoc | 62 ++++++++++
.../parquet/avro/ParquetAvroDataFormat.java | 125 +++++++++++++++++++++
.../parquet/avro/ParquetInputStream.java | 73 ++++++++++++
.../parquet/avro/ParquetOutputStream.java | 94 ++++++++++++++++
.../parquet/avro/ParquetAvroDataFormatTest.java | 61 ++++++++++
...ParquetAvroDataFormatUnmarshalFromFileTest.java | 76 +++++++++++++
.../apache/camel/dataformat/parquet/avro/Pojo.java | 65 +++++++++++
.../avro/SpringParquetAvroDataFormatRouteTest.java | 54 +++++++++
.../SpringParquetAvroDataFormatRouteTest.xml | 45 ++++++++
.../src/test/resources/example1.parquet | Bin 0 -> 793 bytes
.../src/test/resources/log4j2.properties | 27 +++++
components/pom.xml | 1 +
.../services/org/apache/camel/model.properties | 1 +
.../org/apache/camel/model/dataformat/jaxb.index | 1 +
.../apache/camel/model/dataformat/parquetAvro.json | 18 +++
.../resources/org/apache/camel/model/marshal.json | 2 +-
.../org/apache/camel/model/unmarshal.json | 2 +-
.../camel/builder/DataFormatBuilderFactory.java | 8 ++
.../org/apache/camel/builder/DataFormatClause.java | 23 ++++
.../org/apache/camel/model/MarshalDefinition.java | 2 +
.../apache/camel/model/UnmarshalDefinition.java | 2 +
.../model/dataformat/ParquetAvroDataFormat.java | 112 ++++++++++++++++++
.../DataFormatTransformerDefinition.java | 2 +
.../reifier/dataformat/DataFormatReifier.java | 3 +
.../dataformat/ParquetAvroDataFormatReifier.java | 36 ++++++
.../java/org/apache/camel/xml/in/ModelParser.java | 10 ++
.../java/org/apache/camel/xml/out/ModelWriter.java | 17 +++
dsl/camel-endpointdsl/pom.xml | 5 +
.../dsl/yaml/deserializers/ModelDeserializers.java | 69 ++++++++++++
.../deserializers/ModelDeserializersResolver.java | 3 +
.../generated/resources/schema/camel-yaml-dsl.json | 24 ++++
.../generated/resources/schema/camelYamlDsl.json | 24 ++++
parent/pom.xml | 7 ++
43 files changed, 1215 insertions(+), 5 deletions(-)
diff --git a/camel-dependencies/pom.xml b/camel-dependencies/pom.xml
index 830d06e74b3..b2643a02f2d 100644
--- a/camel-dependencies/pom.xml
+++ b/camel-dependencies/pom.xml
@@ -408,6 +408,8 @@
<optaplanner-version>9.37.0.Final</optaplanner-version>
<os-maven-plugin-version>1.7.0</os-maven-plugin-version>
<paho-version>1.2.5</paho-version>
+ <parquet-common-version>1.12.0</parquet-common-version>
+ <parquet-avro-version>1.12.3</parquet-avro-version>
<pdfbox-version>2.0.28</pdfbox-version>
<pgjdbc-driver-version>42.6.0</pgjdbc-driver-version>
<pgjdbc-ng-driver-version>0.8.9</pgjdbc-ng-driver-version>
diff --git a/catalog/camel-allcomponents/pom.xml b/catalog/camel-allcomponents/pom.xml
index f9e02755822..6daf7aff600 100644
--- a/catalog/camel-allcomponents/pom.xml
+++ b/catalog/camel-allcomponents/pom.xml
@@ -1268,6 +1268,11 @@
<artifactId>camel-paho-mqtt5</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-parquet-avro</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-pdf</artifactId>
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models.properties b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models.properties
index ba0d5ba5d26..57d98feccb6 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models.properties
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models.properties
@@ -105,6 +105,7 @@ otherwise
outputType
packageScan
param
+parquetAvro
passThroughServiceFilter
patch
pausable
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/marshal.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/marshal.json
index 377590ec345..b463f73b7ee 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/marshal.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/marshal.json
@@ -12,7 +12,7 @@
"output": false
},
"properties": {
- "dataFormatType": { "index": 0, "kind": "element", "displayName": "Data Format Type", "required": true, "type": "object", "javaType": "org.apache.camel.model.DataFormatDefinition", "oneOf": [ "asn1", "avro", "barcode", "base64", "bindy", "cbor", "crypto", "csv", "custom", "fhirJson", "fhirXml", "flatpack", "grok", "gzipDeflater", "hl7", "ical", "jacksonXml", "jaxb", "json", "jsonApi", "lzf", "mimeMultipart", "pgp", "protobuf", "rss", "soap", "swiftMt", "swiftMx", "syslog", "tarFile", [...]
+ "dataFormatType": { "index": 0, "kind": "element", "displayName": "Data Format Type", "required": true, "type": "object", "javaType": "org.apache.camel.model.DataFormatDefinition", "oneOf": [ "asn1", "avro", "barcode", "base64", "bindy", "cbor", "crypto", "csv", "custom", "fhirJson", "fhirXml", "flatpack", "grok", "gzipDeflater", "hl7", "ical", "jacksonXml", "jaxb", "json", "jsonApi", "lzf", "mimeMultipart", "parquetAvro", "pgp", "protobuf", "rss", "soap", "swiftMt", "swiftMx", "sysl [...]
"disabled": { "index": 1, "kind": "attribute", "displayName": "Disabled", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to disable this EIP from the route during build time. Once an EIP has been disabled then it cannot be enabled later at runtime." },
"id": { "index": 2, "kind": "attribute", "displayName": "Id", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the id of this node" },
"description": { "index": 3, "kind": "element", "displayName": "Description", "required": false, "type": "object", "javaType": "org.apache.camel.model.DescriptionDefinition", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the description of this node" }
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/parquetAvro.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/parquetAvro.json
new file mode 100644
index 00000000000..9ddbb89eeae
--- /dev/null
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/parquetAvro.json
@@ -0,0 +1,18 @@
+{
+ "model": {
+ "kind": "model",
+ "name": "parquetAvro",
+ "title": "Parquet File",
+ "deprecated": false,
+ "firstVersion": "4.0.0",
+ "label": "dataformat,transformation,file",
+ "javaType": "org.apache.camel.model.dataformat.ParquetAvroDataFormat",
+ "abstract": false,
+ "input": false,
+ "output": false
+ },
+ "properties": {
+ "unmarshalType": { "index": 0, "kind": "attribute", "displayName": "Unmarshal Type", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Class to use when unmarshalling." },
+ "id": { "index": 1, "kind": "attribute", "displayName": "Id", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The id of this node" }
+ }
+}
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/unmarshal.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/unmarshal.json
index 115dbcec8f8..a32ed36b462 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/unmarshal.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/unmarshal.json
@@ -12,7 +12,7 @@
"output": false
},
"properties": {
- "dataFormatType": { "index": 0, "kind": "element", "displayName": "Data Format Type", "required": true, "type": "object", "javaType": "org.apache.camel.model.DataFormatDefinition", "oneOf": [ "asn1", "avro", "barcode", "base64", "bindy", "cbor", "crypto", "csv", "custom", "fhirJson", "fhirXml", "flatpack", "grok", "gzipDeflater", "hl7", "ical", "jacksonXml", "jaxb", "json", "jsonApi", "lzf", "mimeMultipart", "pgp", "protobuf", "rss", "soap", "swiftMt", "swiftMx", "syslog", "tarFile", [...]
+ "dataFormatType": { "index": 0, "kind": "element", "displayName": "Data Format Type", "required": true, "type": "object", "javaType": "org.apache.camel.model.DataFormatDefinition", "oneOf": [ "asn1", "avro", "barcode", "base64", "bindy", "cbor", "crypto", "csv", "custom", "fhirJson", "fhirXml", "flatpack", "grok", "gzipDeflater", "hl7", "ical", "jacksonXml", "jaxb", "json", "jsonApi", "lzf", "mimeMultipart", "parquetAvro", "pgp", "protobuf", "rss", "soap", "swiftMt", "swiftMx", "sysl [...]
"allowNullBody": { "index": 1, "kind": "attribute", "displayName": "Allow Null Body", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Indicates whether null is allowed as value of a body to unmarshall." },
"disabled": { "index": 2, "kind": "attribute", "displayName": "Disabled", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to disable this EIP from the route during build time. Once an EIP has been disabled then it cannot be enabled later at runtime." },
"id": { "index": 3, "kind": "attribute", "displayName": "Id", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the id of this node" },
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
index 2f68b25b8aa..8b361a45d76 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
@@ -403,7 +403,7 @@ Enriches a message with data from a secondary resource
<xs:element name="errorHandler" nillable="true" type="xs:anyType">
<xs:annotation>
<xs:documentation xml:lang="en"><![CDATA[
-Camel error handling.
+Error handler settings
]]></xs:documentation>
</xs:annotation>
</xs:element>
@@ -944,6 +944,14 @@ To specify the rest operation parameters.
</xs:annotation>
</xs:element>
+ <xs:element name="parquetAvro" type="tns:parquetAvroDataFormat">
+ <xs:annotation>
+ <xs:documentation xml:lang="en"><![CDATA[
+null
+ ]]></xs:documentation>
+ </xs:annotation>
+ </xs:element>
+
<xs:element name="passThroughServiceFilter" type="tns:passThroughServiceCallServiceFilterConfiguration">
<xs:annotation>
<xs:documentation xml:lang="en"><![CDATA[
@@ -8302,6 +8310,8 @@ it reaches the end when Camel is shut down. Default value: false
<xs:element ref="tns:mimeMultipart"/>
+ <xs:element ref="tns:parquetAvro"/>
+
<xs:element ref="tns:protobuf"/>
<xs:element ref="tns:rss"/>
@@ -10357,6 +10367,28 @@ Defines whether the content of binary parts in the MIME multipart is binary
</xs:complexType>
+ <xs:complexType name="parquetAvroDataFormat">
+
+ <xs:complexContent>
+
+ <xs:extension base="tns:dataFormat">
+
+ <xs:sequence/>
+
+ <xs:attribute name="unmarshalType" type="xs:string">
+ <xs:annotation>
+ <xs:documentation xml:lang="en"><![CDATA[
+Class to use when unmarshalling.
+ ]]></xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
+
+ </xs:extension>
+
+ </xs:complexContent>
+
+ </xs:complexType>
+
<xs:complexType name="protobufDataFormat">
<xs:complexContent>
@@ -16623,6 +16655,8 @@ Sets a reference to use for lookup the policy in the registry.
<xs:element ref="tns:mimeMultipart"/>
+ <xs:element ref="tns:parquetAvro"/>
+
<xs:element ref="tns:protobuf"/>
<xs:element ref="tns:rss"/>
@@ -20077,6 +20111,8 @@ to refer to an existing data format instance.
<xs:element ref="tns:mimeMultipart"/>
+ <xs:element ref="tns:parquetAvro"/>
+
<xs:element ref="tns:protobuf"/>
<xs:element ref="tns:rss"/>
diff --git a/components/camel-parquet-avro/pom.xml b/components/camel-parquet-avro/pom.xml
new file mode 100644
index 00000000000..1e0bb0e595c
--- /dev/null
+++ b/components/camel-parquet-avro/pom.xml
@@ -0,0 +1,89 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>components</artifactId>
+ <groupId>org.apache.camel</groupId>
+ <version>4.0.0-SNAPSHOT</version>
+ </parent>
+
+ <groupId>org.apache.camel.dataformat.parquet.avro</groupId>
+ <artifactId>camel-parquet-avro</artifactId>
+ <packaging>jar</packaging>
+ <version>4.0.0-SNAPSHOT</version>
+
+ <name>Camel :: ParquetAvro</name>
+ <description>Camel ParquetAvro DataFormat</description>
+
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-support</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <version>${commons-io-version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-common</artifactId>
+ <version>${parquet-common-version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ <version>${hadoop3-version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-avro</artifactId>
+ <version>${parquet-avro-version}</version>
+ </dependency>
+
+ <!-- logging -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <version>${log4j2-version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- testing -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-junit5</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-spring-xml</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-spring-junit5</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>jakarta.xml.bind</groupId>
+ <artifactId>jakarta.xml.bind-api</artifactId>
+ <version>${jakarta-xml-bind-api-version}</version>
+ <optional>true</optional>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/components/camel-parquet-avro/src/generated/resources/META-INF/services/org/apache/camel/dataformat.properties b/components/camel-parquet-avro/src/generated/resources/META-INF/services/org/apache/camel/dataformat.properties
new file mode 100644
index 00000000000..b7957ac1a85
--- /dev/null
+++ b/components/camel-parquet-avro/src/generated/resources/META-INF/services/org/apache/camel/dataformat.properties
@@ -0,0 +1,7 @@
+# Generated by camel build tools - do NOT edit this file!
+dataFormats=parquetAvro
+groupId=org.apache.camel.dataformat.parquet.avro
+artifactId=camel-parquet-avro
+version=4.0.0-SNAPSHOT
+projectName=Camel :: ParquetAvro
+projectDescription=Camel ParquetAvro DataFormat
diff --git a/components/camel-parquet-avro/src/generated/resources/META-INF/services/org/apache/camel/dataformat/parquetAvro b/components/camel-parquet-avro/src/generated/resources/META-INF/services/org/apache/camel/dataformat/parquetAvro
new file mode 100644
index 00000000000..ec4060cff52
--- /dev/null
+++ b/components/camel-parquet-avro/src/generated/resources/META-INF/services/org/apache/camel/dataformat/parquetAvro
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.dataformat.parquet.avro.ParquetAvroDataFormat
diff --git a/components/camel-parquet-avro/src/main/doc/parquetAvro-dataformat.adoc b/components/camel-parquet-avro/src/main/doc/parquetAvro-dataformat.adoc
new file mode 100644
index 00000000000..54e21dc4f04
--- /dev/null
+++ b/components/camel-parquet-avro/src/main/doc/parquetAvro-dataformat.adoc
@@ -0,0 +1,62 @@
+= ParquetAvro File DataFormat
+:doctitle: ParquetAvro Format
+:shortname: parquetAvro
+:artifactid: camel-parquet-avro
+:description: Unmarshal and marshal parquet format using avro (de)/serializers
+:since: 4.0
+:supportlevel: Unstable
+:tabs-sync-option:
+//Manually maintained attributes
+:camel-spring-boot-name: parquet-avro
+
+*Since Camel {since}*
+
+The ParquetAvro Data Format Data Format is a Camel Frameworks's data format implementation based on parquet-avro library for (de)/serialization purposes. Messages can be unmarshalled (conversion to simple Java POJO(s)) to plain Java objects. By the help of Camel's routing engine and data transformations you can then play with POJO(s) and apply customised formatting and call other Camel Component's to convert and send messages to upstream systems.
+
+== Parquet Data Format Options
+
+// dataformat options: START
+include::partial$dataformat-options.adoc[]
+// dataformat options: END
+
+== Unmarshal
+
+There are ways to unmarshal parquet files/structures (Usually binary parquet files) where camel DSL allows
+
+In this first example we unmarshal file payload to OutputStream and send it to mock endpoint, then we will be able to get Pojo (it could be a list if that is coming through)
+
+[source,java]
+-----------------------------------------------------------------------
+from("direct:unmarshal").unmarshal(parquet).to("mock:unmarshal");
+-----------------------------------------------------------------------
+
+== Marshal
+
+Marshalling is the reverse process of unmarshalling so when you have your Pojo and marshal it you will get the parquet formatted output stream on your producer endpoint.
+
+[source,java]
+-----------------------------------------------------------------------
+from("direct:marshal").marshal(parquet).to("mock:marshal");
+-----------------------------------------------------------------------
+
+== Dependencies
+
+To use parquet-avro data format in your camel routes you need to add a dependency on
+*camel-parquet-avro* which implements this data format.
+
+If you use Maven you can just add the following to your `pom.xml`,
+substituting the version number for the latest & greatest release (see
+the download page for the latest versions).
+
+[source,xml]
+----------------------------------------------------------
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-parquet-avro</artifactId>
+ <version>x.x.x</version>
+ <!-- use the same version as your Camel core version -->
+</dependency>
+----------------------------------------------------------
+
+
+include::spring-boot:partial$starter.adoc[]
\ No newline at end of file
diff --git a/components/camel-parquet-avro/src/main/java/org/apache/camel/dataformat/parquet/avro/ParquetAvroDataFormat.java b/components/camel-parquet-avro/src/main/java/org/apache/camel/dataformat/parquet/avro/ParquetAvroDataFormat.java
new file mode 100644
index 00000000000..ec5496bb62b
--- /dev/null
+++ b/components/camel-parquet-avro/src/main/java/org/apache/camel/dataformat/parquet/avro/ParquetAvroDataFormat.java
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.dataformat.parquet.avro;
+
+import java.io.BufferedOutputStream;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.avro.reflect.ReflectData;
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.DataFormat;
+import org.apache.camel.spi.DataFormatName;
+import org.apache.camel.spi.annotations.Dataformat;
+import org.apache.camel.support.DefaultUuidGenerator;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.parquet.avro.AvroParquetReader;
+import org.apache.parquet.avro.AvroParquetWriter;
+import org.apache.parquet.hadoop.ParquetReader;
+import org.apache.parquet.hadoop.ParquetWriter;
+
+import static org.apache.parquet.hadoop.ParquetFileWriter.Mode.OVERWRITE;
+import static org.apache.parquet.hadoop.metadata.CompressionCodecName.GZIP;
+
+@Dataformat("parquetAvro")
+public class ParquetAvroDataFormat extends ServiceSupport implements DataFormat, DataFormatName {
+
+ private static final DefaultUuidGenerator DEFAULT_UUID_GENERATOR = new DefaultUuidGenerator();
+
+ private Class<?> unmarshalType;
+
+ public String getDataFormatName() {
+ return "parquet-avro";
+ }
+
+ public void marshal(Exchange exchange, Object graph, OutputStream stream) throws Exception {
+ // marshal from the Java object (graph) to the parquet-avro type
+ Configuration conf = new Configuration();
+
+ FileSystem.get(conf).setWriteChecksum(false);
+
+ BufferedOutputStream parquetOutput = new BufferedOutputStream(stream);
+ ParquetOutputStream parquetOutputStream = new ParquetOutputStream(
+ DEFAULT_UUID_GENERATOR.generateUuid(),
+ parquetOutput);
+
+ List<?> list = (List<?>) graph;
+
+ try (ParquetWriter writer = AvroParquetWriter.builder(parquetOutputStream)
+ .withSchema(ReflectData.AllowNull.get().getSchema(unmarshalType)) // generate nullable fields
+ .withDataModel(ReflectData.get())
+ .withConf(conf)
+ .withCompressionCodec(GZIP)
+ .withWriteMode(OVERWRITE)
+ .build()) {
+ for (Object grapElem : list) {
+ writer.write(grapElem);
+ }
+ }
+ }
+
+ public Object unmarshal(Exchange exchange, InputStream stream) throws Exception {
+ // unmarshal from the input stream of parquet-avro to Java object (graph)
+ List<Object> parquetObjects = new ArrayList<>();
+ Configuration conf = new Configuration();
+
+ ParquetInputStream parquetInputStream = new ParquetInputStream(
+ DEFAULT_UUID_GENERATOR.generateUuid(),
+ stream.readAllBytes());
+
+ ParquetReader r = AvroParquetReader.builder(parquetInputStream)
+ .disableCompatibility()
+ .build();
+
+ try (ParquetReader reader = AvroParquetReader.builder(parquetInputStream)
+ .withDataModel(new ReflectData(unmarshalType.getClassLoader()))
+ .disableCompatibility() // always use this (since this is a new project)
+ .withConf(conf)
+ .build()) {
+
+ Object pojo;
+ while ((pojo = unmarshalType.cast(reader.read())) != null) {
+ parquetObjects.add(pojo);
+ }
+ }
+
+ return parquetObjects;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ // no-op
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ // no-op
+ }
+
+ public Class<?> getUnmarshalType() {
+ return unmarshalType;
+ }
+
+ public void setUnmarshalType(Class<?> unmarshalType) {
+ this.unmarshalType = unmarshalType;
+ }
+
+}
diff --git a/components/camel-parquet-avro/src/main/java/org/apache/camel/dataformat/parquet/avro/ParquetInputStream.java b/components/camel-parquet-avro/src/main/java/org/apache/camel/dataformat/parquet/avro/ParquetInputStream.java
new file mode 100644
index 00000000000..b65f976f982
--- /dev/null
+++ b/components/camel-parquet-avro/src/main/java/org/apache/camel/dataformat/parquet/avro/ParquetInputStream.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.dataformat.parquet.avro;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+
+import org.apache.parquet.io.DelegatingSeekableInputStream;
+import org.apache.parquet.io.InputFile;
+import org.apache.parquet.io.SeekableInputStream;
+
+public class ParquetInputStream implements InputFile {
+ private final String streamId;
+ private final byte[] data;
+
+ private static class SeekableByteArrayInputStream extends ByteArrayInputStream {
+ public SeekableByteArrayInputStream(byte[] buf) {
+ super(buf);
+ }
+
+ public void setPos(int pos) {
+ this.pos = pos;
+ }
+
+ public int getPos() {
+ return this.pos;
+ }
+ }
+
+ public ParquetInputStream(String streamId, byte[] data) {
+ this.streamId = streamId;
+ this.data = data;
+ }
+
+ @Override
+ public long getLength() throws IOException {
+ return this.data.length;
+ }
+
+ @Override
+ public SeekableInputStream newStream() throws IOException {
+ return new DelegatingSeekableInputStream(new SeekableByteArrayInputStream(this.data)) {
+ @Override
+ public void seek(long newPos) throws IOException {
+ ((SeekableByteArrayInputStream) this.getStream()).setPos((int) newPos);
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return ((SeekableByteArrayInputStream) this.getStream()).getPos();
+ }
+ };
+ }
+
+ @Override
+ public String toString() {
+ return "ParquetInputStream[" + streamId + "]";
+ }
+}
diff --git a/components/camel-parquet-avro/src/main/java/org/apache/camel/dataformat/parquet/avro/ParquetOutputStream.java b/components/camel-parquet-avro/src/main/java/org/apache/camel/dataformat/parquet/avro/ParquetOutputStream.java
new file mode 100644
index 00000000000..c3f630f7f78
--- /dev/null
+++ b/components/camel-parquet-avro/src/main/java/org/apache/camel/dataformat/parquet/avro/ParquetOutputStream.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.dataformat.parquet.avro;
+
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+
+import org.apache.parquet.io.OutputFile;
+import org.apache.parquet.io.PositionOutputStream;
+
+public class ParquetOutputStream implements OutputFile {
+
+ private final String streamId;
+ private final BufferedOutputStream out;
+
+ private int pos;
+
+ public ParquetOutputStream(String streamId, BufferedOutputStream out) {
+ this.pos = 0;
+ this.streamId = streamId;
+ this.out = out;
+ }
+
+ @Override
+ public PositionOutputStream create(long blockSizeHint) throws IOException {
+ return createPositionOutputstream();
+ }
+
+ private PositionOutputStream createPositionOutputstream() {
+ return new PositionOutputStream() {
+
+ @Override
+ public long getPos() throws IOException {
+ return pos;
+ }
+
+ @Override
+ public void flush() throws IOException {
+ out.flush();
+ };
+
+ @Override
+ public void close() throws IOException {
+ out.close();
+ };
+
+ @Override
+ public void write(int b) throws IOException {
+ out.write(b);
+ pos++;
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ out.write(b, off, len);
+ pos += len;
+ }
+ };
+ }
+
+ @Override
+ public PositionOutputStream createOrOverwrite(long blockSizeHint) throws IOException {
+ return createPositionOutputstream();
+ }
+
+ @Override
+ public boolean supportsBlockSize() {
+ return false;
+ }
+
+ @Override
+ public long defaultBlockSize() {
+ return 0;
+ }
+
+ @Override
+ public String toString() {
+ return "ParquetOutputStream[" + streamId + "]";
+ }
+}
diff --git a/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/ParquetAvroDataFormatTest.java b/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/ParquetAvroDataFormatTest.java
new file mode 100644
index 00000000000..3c086622451
--- /dev/null
+++ b/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/ParquetAvroDataFormatTest.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.dataformat.parquet.avro;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+
+public class ParquetAvroDataFormatTest extends CamelTestSupport {
+
+ @Test
+ public void testMarshalAndUnmarshalMap() throws Exception {
+ Collection<Pojo> in = List.of(
+ new Pojo(1, "airport"),
+ new Pojo(2, "penguin"),
+ new Pojo(3, "verb"));
+ MockEndpoint mock = getMockEndpoint("mock:reverse");
+ mock.message(0).body().isEqualTo(in);
+
+ Object marshalled = template.requestBody("direct:in", in);
+ template.sendBody("direct:back", marshalled);
+ mock.assertIsSatisfied();
+ List<Exchange> receivedExchanges = mock.getReceivedExchanges();
+ assertArrayEquals(in.toArray(), receivedExchanges.get(0).getIn().getBody(List.class).toArray());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() throws Exception {
+ ParquetAvroDataFormat format = new ParquetAvroDataFormat();
+ format.setUnmarshalType(Pojo.class);
+ from("direct:in").marshal(format);
+ from("direct:back").unmarshal(format).to("mock:reverse");
+ }
+ };
+ }
+}
diff --git a/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/ParquetAvroDataFormatUnmarshalFromFileTest.java b/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/ParquetAvroDataFormatUnmarshalFromFileTest.java
new file mode 100644
index 00000000000..d83586f7d0f
--- /dev/null
+++ b/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/ParquetAvroDataFormatUnmarshalFromFileTest.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.dataformat.parquet.avro;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.apache.commons.io.FileUtils;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ParquetAvroDataFormatUnmarshalFromFileTest extends CamelTestSupport {
+
+ @Test
+ public void testUnmarshalFromFile() throws Exception {
+ Collection<Pojo> in = List.of(
+ new Pojo(1, "airport"),
+ new Pojo(2, "penguin"),
+ new Pojo(3, "verb"));
+ getMockEndpoint("mock:reverse").expectedMessageCount(1);
+ File testFile = new File("src/test/resources/example1.parquet");
+ ByteArrayInputStream bais = reteriveByteArrayInputStream(testFile);
+
+ template.sendBody("direct:fromFile", bais);
+
+ List<Exchange> exchanges = getMockEndpoint("mock:reverse").getExchanges();
+
+ assertEquals(1, exchanges.size());
+ for (Exchange exchange : exchanges) {
+ List out = exchange.getIn().getBody(List.class);
+ assertArrayEquals(in.toArray(), out.toArray());
+ }
+
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ private ByteArrayInputStream reteriveByteArrayInputStream(File file) throws IOException {
+ return new ByteArrayInputStream(FileUtils.readFileToByteArray(file));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() throws Exception {
+ ParquetAvroDataFormat format = new ParquetAvroDataFormat();
+ format.setUnmarshalType(Pojo.class);
+ from("direct:fromFile").unmarshal(format).to("mock:reverse");
+ }
+ };
+ }
+}
diff --git a/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/Pojo.java b/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/Pojo.java
new file mode 100644
index 00000000000..c5a3771c250
--- /dev/null
+++ b/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/Pojo.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.dataformat.parquet.avro;
+
+public class Pojo {
+ private long id;
+ private String data;
+
+ public Pojo() {
+ }
+
+ public Pojo(long id, String data) {
+ this.id = id;
+ this.data = data;
+ }
+
+ public long getId() {
+ return id;
+ }
+
+ public String getData() {
+ return data;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("Pojo[");
+ sb.append("id=").append(id);
+ sb.append(", data=").append(data);
+ sb.append("]");
+ return sb.toString();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Pojo pojo = (Pojo) o;
+ return id == pojo.id && java.util.Objects.equals(data, pojo.data);
+ }
+
+ @Override
+ public int hashCode() {
+ return java.util.Objects.hash(id, data);
+ }
+}
diff --git a/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/SpringParquetAvroDataFormatRouteTest.java b/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/SpringParquetAvroDataFormatRouteTest.java
new file mode 100644
index 00000000000..5f30d43e80e
--- /dev/null
+++ b/components/camel-parquet-avro/src/test/java/org/apache/camel/dataformat/parquet/avro/SpringParquetAvroDataFormatRouteTest.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.dataformat.parquet.avro;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.spring.junit5.CamelSpringTestSupport;
+import org.junit.jupiter.api.Test;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+
+public class SpringParquetAvroDataFormatRouteTest extends CamelSpringTestSupport {
+
+ @Test
+ public void testSpringMarshalAndUnmarshalMap() throws Exception {
+ Collection<Pojo> in = List.of(
+ new Pojo(1, "airport"),
+ new Pojo(2, "penguin"),
+ new Pojo(3, "verb"));
+ MockEndpoint mock = getMockEndpoint("mock:reverse");
+ mock.message(0).body().isEqualTo(in);
+
+ Object marshalled = template.requestBody("direct:marshal", in);
+ template.sendBody("direct:unmarshal", marshalled);
+ mock.assertIsSatisfied();
+ List<Exchange> receivedExchanges = mock.getReceivedExchanges();
+ assertArrayEquals(in.toArray(), receivedExchanges.get(0).getIn().getBody(List.class).toArray());
+ }
+
+ @Override
+ protected AbstractApplicationContext createApplicationContext() {
+ return new ClassPathXmlApplicationContext("SpringParquetAvroDataFormatRouteTest.xml");
+ }
+
+}
diff --git a/components/camel-parquet-avro/src/test/resources/SpringParquetAvroDataFormatRouteTest.xml b/components/camel-parquet-avro/src/test/resources/SpringParquetAvroDataFormatRouteTest.xml
new file mode 100644
index 00000000000..42b91142434
--- /dev/null
+++ b/components/camel-parquet-avro/src/test/resources/SpringParquetAvroDataFormatRouteTest.xml
@@ -0,0 +1,45 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<beans xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="
+ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+ http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
+ ">
+
+ <!-- START SNIPPET: e1 -->
+ <camelContext xmlns="http://camel.apache.org/schema/spring">
+ <route>
+ <from uri="direct:marshal"/>
+ <marshal>
+ <parquetAvro unmarshalType="org.apache.camel.dataformat.parquet.avro.Pojo" />
+ </marshal>
+ </route>
+ <route>
+ <from uri="direct:unmarshal"/>
+ <unmarshal>
+ <parquetAvro unmarshalType="org.apache.camel.dataformat.parquet.avro.Pojo" />
+ </unmarshal>
+ <to uri="mock:reverse"/>
+ </route>
+ </camelContext>
+ <!-- END SNIPPET: e1 -->
+
+</beans>
diff --git a/components/camel-parquet-avro/src/test/resources/example1.parquet b/components/camel-parquet-avro/src/test/resources/example1.parquet
new file mode 100644
index 00000000000..d657e8b6d60
Binary files /dev/null and b/components/camel-parquet-avro/src/test/resources/example1.parquet differ
diff --git a/components/camel-parquet-avro/src/test/resources/log4j2.properties b/components/camel-parquet-avro/src/test/resources/log4j2.properties
new file mode 100644
index 00000000000..25786440746
--- /dev/null
+++ b/components/camel-parquet-avro/src/test/resources/log4j2.properties
@@ -0,0 +1,27 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ---------------------------------------------------------------------------
+appender.file.type = File
+appender.file.name = file
+appender.file.fileName = target/camel-parquet-avro-test.log
+appender.file.layout.type = PatternLayout
+appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+appender.out.type = Console
+appender.out.name = out
+appender.out.layout.type = PatternLayout
+appender.out.layout.pattern = [%30.30t] %-30.30c{1} %-5p %m%n
+rootLogger.level = INFO
+rootLogger.appenderRef.file.ref = file
\ No newline at end of file
diff --git a/components/pom.xml b/components/pom.xml
index 70e7d8b091b..7ab55270ae1 100644
--- a/components/pom.xml
+++ b/components/pom.xml
@@ -229,6 +229,7 @@
<module>camel-opentelemetry</module>
<module>camel-paho</module>
<module>camel-paho-mqtt5</module>
+ <module>camel-parquet-avro</module>
<module>camel-pdf</module>
<module>camel-pg-replication-slot</module>
<module>camel-pgevent</module>
diff --git a/core/camel-core-model/src/generated/resources/META-INF/services/org/apache/camel/model.properties b/core/camel-core-model/src/generated/resources/META-INF/services/org/apache/camel/model.properties
index 8b2374be646..2125f2df0aa 100644
--- a/core/camel-core-model/src/generated/resources/META-INF/services/org/apache/camel/model.properties
+++ b/core/camel-core-model/src/generated/resources/META-INF/services/org/apache/camel/model.properties
@@ -106,6 +106,7 @@ otherwise
outputType
packageScan
param
+parquetAvro
passThroughServiceFilter
patch
pausable
diff --git a/core/camel-core-model/src/generated/resources/org/apache/camel/model/dataformat/jaxb.index b/core/camel-core-model/src/generated/resources/org/apache/camel/model/dataformat/jaxb.index
index f407397532d..7b62528f186 100644
--- a/core/camel-core-model/src/generated/resources/org/apache/camel/model/dataformat/jaxb.index
+++ b/core/camel-core-model/src/generated/resources/org/apache/camel/model/dataformat/jaxb.index
@@ -26,6 +26,7 @@ JsonLibrary
LZFDataFormat
MimeMultipartDataFormat
PGPDataFormat
+ParquetAvroDataFormat
ProtobufDataFormat
ProtobufLibrary
RssDataFormat
diff --git a/core/camel-core-model/src/generated/resources/org/apache/camel/model/dataformat/parquetAvro.json b/core/camel-core-model/src/generated/resources/org/apache/camel/model/dataformat/parquetAvro.json
new file mode 100644
index 00000000000..9ddbb89eeae
--- /dev/null
+++ b/core/camel-core-model/src/generated/resources/org/apache/camel/model/dataformat/parquetAvro.json
@@ -0,0 +1,18 @@
+{
+ "model": {
+ "kind": "model",
+ "name": "parquetAvro",
+ "title": "Parquet File",
+ "deprecated": false,
+ "firstVersion": "4.0.0",
+ "label": "dataformat,transformation,file",
+ "javaType": "org.apache.camel.model.dataformat.ParquetAvroDataFormat",
+ "abstract": false,
+ "input": false,
+ "output": false
+ },
+ "properties": {
+ "unmarshalType": { "index": 0, "kind": "attribute", "displayName": "Unmarshal Type", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Class to use when unmarshalling." },
+ "id": { "index": 1, "kind": "attribute", "displayName": "Id", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The id of this node" }
+ }
+}
diff --git a/core/camel-core-model/src/generated/resources/org/apache/camel/model/marshal.json b/core/camel-core-model/src/generated/resources/org/apache/camel/model/marshal.json
index 377590ec345..b463f73b7ee 100644
--- a/core/camel-core-model/src/generated/resources/org/apache/camel/model/marshal.json
+++ b/core/camel-core-model/src/generated/resources/org/apache/camel/model/marshal.json
@@ -12,7 +12,7 @@
"output": false
},
"properties": {
- "dataFormatType": { "index": 0, "kind": "element", "displayName": "Data Format Type", "required": true, "type": "object", "javaType": "org.apache.camel.model.DataFormatDefinition", "oneOf": [ "asn1", "avro", "barcode", "base64", "bindy", "cbor", "crypto", "csv", "custom", "fhirJson", "fhirXml", "flatpack", "grok", "gzipDeflater", "hl7", "ical", "jacksonXml", "jaxb", "json", "jsonApi", "lzf", "mimeMultipart", "pgp", "protobuf", "rss", "soap", "swiftMt", "swiftMx", "syslog", "tarFile", [...]
+ "dataFormatType": { "index": 0, "kind": "element", "displayName": "Data Format Type", "required": true, "type": "object", "javaType": "org.apache.camel.model.DataFormatDefinition", "oneOf": [ "asn1", "avro", "barcode", "base64", "bindy", "cbor", "crypto", "csv", "custom", "fhirJson", "fhirXml", "flatpack", "grok", "gzipDeflater", "hl7", "ical", "jacksonXml", "jaxb", "json", "jsonApi", "lzf", "mimeMultipart", "parquetAvro", "pgp", "protobuf", "rss", "soap", "swiftMt", "swiftMx", "sysl [...]
"disabled": { "index": 1, "kind": "attribute", "displayName": "Disabled", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to disable this EIP from the route during build time. Once an EIP has been disabled then it cannot be enabled later at runtime." },
"id": { "index": 2, "kind": "attribute", "displayName": "Id", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the id of this node" },
"description": { "index": 3, "kind": "element", "displayName": "Description", "required": false, "type": "object", "javaType": "org.apache.camel.model.DescriptionDefinition", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the description of this node" }
diff --git a/core/camel-core-model/src/generated/resources/org/apache/camel/model/unmarshal.json b/core/camel-core-model/src/generated/resources/org/apache/camel/model/unmarshal.json
index 115dbcec8f8..a32ed36b462 100644
--- a/core/camel-core-model/src/generated/resources/org/apache/camel/model/unmarshal.json
+++ b/core/camel-core-model/src/generated/resources/org/apache/camel/model/unmarshal.json
@@ -12,7 +12,7 @@
"output": false
},
"properties": {
- "dataFormatType": { "index": 0, "kind": "element", "displayName": "Data Format Type", "required": true, "type": "object", "javaType": "org.apache.camel.model.DataFormatDefinition", "oneOf": [ "asn1", "avro", "barcode", "base64", "bindy", "cbor", "crypto", "csv", "custom", "fhirJson", "fhirXml", "flatpack", "grok", "gzipDeflater", "hl7", "ical", "jacksonXml", "jaxb", "json", "jsonApi", "lzf", "mimeMultipart", "pgp", "protobuf", "rss", "soap", "swiftMt", "swiftMx", "syslog", "tarFile", [...]
+ "dataFormatType": { "index": 0, "kind": "element", "displayName": "Data Format Type", "required": true, "type": "object", "javaType": "org.apache.camel.model.DataFormatDefinition", "oneOf": [ "asn1", "avro", "barcode", "base64", "bindy", "cbor", "crypto", "csv", "custom", "fhirJson", "fhirXml", "flatpack", "grok", "gzipDeflater", "hl7", "ical", "jacksonXml", "jaxb", "json", "jsonApi", "lzf", "mimeMultipart", "parquetAvro", "pgp", "protobuf", "rss", "soap", "swiftMt", "swiftMx", "sysl [...]
"allowNullBody": { "index": 1, "kind": "attribute", "displayName": "Allow Null Body", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Indicates whether null is allowed as value of a body to unmarshall." },
"disabled": { "index": 2, "kind": "attribute", "displayName": "Disabled", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to disable this EIP from the route during build time. Once an EIP has been disabled then it cannot be enabled later at runtime." },
"id": { "index": 3, "kind": "attribute", "displayName": "Id", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the id of this node" },
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/builder/DataFormatBuilderFactory.java b/core/camel-core-model/src/main/java/org/apache/camel/builder/DataFormatBuilderFactory.java
index 737ef0dab9d..8c1072e35b3 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/builder/DataFormatBuilderFactory.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/builder/DataFormatBuilderFactory.java
@@ -39,6 +39,7 @@ import org.apache.camel.model.dataformat.JsonDataFormat;
import org.apache.camel.model.dataformat.LZFDataFormat;
import org.apache.camel.model.dataformat.MimeMultipartDataFormat;
import org.apache.camel.model.dataformat.PGPDataFormat;
+import org.apache.camel.model.dataformat.ParquetAvroDataFormat;
import org.apache.camel.model.dataformat.ProtobufDataFormat;
import org.apache.camel.model.dataformat.RssDataFormat;
import org.apache.camel.model.dataformat.SoapDataFormat;
@@ -215,6 +216,13 @@ public final class DataFormatBuilderFactory {
return new MimeMultipartDataFormat.Builder();
}
+ /**
+ * Uses the protobuf data format
+ */
+ public ParquetAvroDataFormat.Builder parquetAvro() {
+ return new ParquetAvroDataFormat.Builder();
+ }
+
/**
* Uses the PGP data format
*/
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/builder/DataFormatClause.java b/core/camel-core-model/src/main/java/org/apache/camel/builder/DataFormatClause.java
index 945100c211b..59f2e1649e4 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/builder/DataFormatClause.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/builder/DataFormatClause.java
@@ -45,6 +45,7 @@ import org.apache.camel.model.dataformat.JsonLibrary;
import org.apache.camel.model.dataformat.LZFDataFormat;
import org.apache.camel.model.dataformat.MimeMultipartDataFormat;
import org.apache.camel.model.dataformat.PGPDataFormat;
+import org.apache.camel.model.dataformat.ParquetAvroDataFormat;
import org.apache.camel.model.dataformat.ProtobufDataFormat;
import org.apache.camel.model.dataformat.ProtobufLibrary;
import org.apache.camel.model.dataformat.RssDataFormat;
@@ -1251,6 +1252,28 @@ public class DataFormatClause<T extends ProcessorDefinition<?>> {
return dataFormat(new ASN1DataFormat(usingIterator));
}
+ /**
+ * Uses the parquet-avro file data format
+ */
+ public T parquetAvro() {
+ ParquetAvroDataFormat parquetAvroDataFormat = new ParquetAvroDataFormat();
+ return dataFormat(parquetAvroDataFormat);
+ }
+
+ /**
+ * Uses the parquet-avro file data format
+ */
+ public T parquetAvro(String unmarshalType) {
+ return dataFormat(new ParquetAvroDataFormat(unmarshalType));
+ }
+
+ /**
+ * Uses the parquet-avro file data format
+ */
+ public T parquetAvro(Class<?> unmarshalType) {
+ return dataFormat(new ParquetAvroDataFormat(unmarshalType));
+ }
+
/**
* Uses the FHIR JSON data format
*/
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/MarshalDefinition.java b/core/camel-core-model/src/main/java/org/apache/camel/model/MarshalDefinition.java
index f7a57469fef..0e87066c92d 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/model/MarshalDefinition.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/MarshalDefinition.java
@@ -45,6 +45,7 @@ import org.apache.camel.model.dataformat.JsonDataFormat;
import org.apache.camel.model.dataformat.LZFDataFormat;
import org.apache.camel.model.dataformat.MimeMultipartDataFormat;
import org.apache.camel.model.dataformat.PGPDataFormat;
+import org.apache.camel.model.dataformat.ParquetAvroDataFormat;
import org.apache.camel.model.dataformat.ProtobufDataFormat;
import org.apache.camel.model.dataformat.RssDataFormat;
import org.apache.camel.model.dataformat.SoapDataFormat;
@@ -94,6 +95,7 @@ public class MarshalDefinition extends NoOutputDefinition<MarshalDefinition> imp
@XmlElement(name = "jsonApi", type = JsonApiDataFormat.class),
@XmlElement(name = "lzf", type = LZFDataFormat.class),
@XmlElement(name = "mimeMultipart", type = MimeMultipartDataFormat.class),
+ @XmlElement(name = "parquetAvro", type = ParquetAvroDataFormat.class),
@XmlElement(name = "protobuf", type = ProtobufDataFormat.class),
@XmlElement(name = "rss", type = RssDataFormat.class),
@XmlElement(name = "soap", type = SoapDataFormat.class),
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/UnmarshalDefinition.java b/core/camel-core-model/src/main/java/org/apache/camel/model/UnmarshalDefinition.java
index 0c36e6f026a..29a691e6750 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/model/UnmarshalDefinition.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/UnmarshalDefinition.java
@@ -46,6 +46,7 @@ import org.apache.camel.model.dataformat.JsonDataFormat;
import org.apache.camel.model.dataformat.LZFDataFormat;
import org.apache.camel.model.dataformat.MimeMultipartDataFormat;
import org.apache.camel.model.dataformat.PGPDataFormat;
+import org.apache.camel.model.dataformat.ParquetAvroDataFormat;
import org.apache.camel.model.dataformat.ProtobufDataFormat;
import org.apache.camel.model.dataformat.RssDataFormat;
import org.apache.camel.model.dataformat.SoapDataFormat;
@@ -95,6 +96,7 @@ public class UnmarshalDefinition extends NoOutputDefinition<UnmarshalDefinition>
@XmlElement(name = "jsonApi", type = JsonApiDataFormat.class),
@XmlElement(name = "lzf", type = LZFDataFormat.class),
@XmlElement(name = "mimeMultipart", type = MimeMultipartDataFormat.class),
+ @XmlElement(name = "parquetAvro", type = ParquetAvroDataFormat.class),
@XmlElement(name = "protobuf", type = ProtobufDataFormat.class),
@XmlElement(name = "rss", type = RssDataFormat.class),
@XmlElement(name = "soap", type = SoapDataFormat.class),
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/dataformat/ParquetAvroDataFormat.java b/core/camel-core-model/src/main/java/org/apache/camel/model/dataformat/ParquetAvroDataFormat.java
new file mode 100644
index 00000000000..ab784706d03
--- /dev/null
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/dataformat/ParquetAvroDataFormat.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.model.dataformat;
+
+import jakarta.xml.bind.annotation.XmlAccessType;
+import jakarta.xml.bind.annotation.XmlAccessorType;
+import jakarta.xml.bind.annotation.XmlAttribute;
+import jakarta.xml.bind.annotation.XmlRootElement;
+import jakarta.xml.bind.annotation.XmlTransient;
+
+import org.apache.camel.builder.DataFormatBuilder;
+import org.apache.camel.model.DataFormatDefinition;
+import org.apache.camel.spi.Metadata;
+
+@Metadata(firstVersion = "4.0.0", label = "dataformat,transformation,file", title = "Parquet File")
+@XmlRootElement(name = "parquetAvro")
+@XmlAccessorType(XmlAccessType.FIELD)
+public class ParquetAvroDataFormat extends DataFormatDefinition {
+
+ @XmlTransient
+ private Class<?> unmarshalType;
+
+ @XmlAttribute(name = "unmarshalType")
+ private String unmarshalTypeName;
+
+ public ParquetAvroDataFormat() {
+ super("parquetAvro");
+ }
+
+ public ParquetAvroDataFormat(String unmarshalTypeName) {
+ this();
+ setUnmarshalTypeName(unmarshalTypeName);
+ }
+
+ public ParquetAvroDataFormat(Class<?> unmarshalType) {
+ this();
+ this.unmarshalType = unmarshalType;
+ }
+
+ private ParquetAvroDataFormat(Builder builder) {
+ this();
+ this.unmarshalTypeName = builder.unmarshalTypeName;
+ this.unmarshalType = builder.unmarshalType;
+ }
+
+ public Class<?> getUnmarshalType() {
+ return unmarshalType;
+ }
+
+ /**
+ * Class to use when unmarshalling.
+ */
+ public void setUnmarshalType(Class<?> unmarshalType) {
+ this.unmarshalType = unmarshalType;
+ }
+
+ /**
+ * Class to use when unmarshalling.
+ */
+ public void setUnmarshalTypeName(String unmarshalTypeName) {
+ this.unmarshalTypeName = unmarshalTypeName;
+ }
+
+ public String getUnmarshalTypeName() {
+ return unmarshalTypeName;
+ }
+
+ /**
+ * {@code Builder} is a specific builder for {@link ParquetAvroDataFormat}.
+ */
+ @XmlTransient
+ public static class Builder implements DataFormatBuilder<ParquetAvroDataFormat> {
+
+ private Class<?> unmarshalType;
+ private String unmarshalTypeName;
+ /**
+ * Class to use when unmarshalling.
+ */
+ public Builder unmarshalTypeName(String unmarshalTypeName) {
+ this.unmarshalTypeName = unmarshalTypeName;
+ return this;
+ }
+
+ /**
+ * Class to use when unmarshalling.
+ */
+ public Builder unmarshalType(Class<?> unmarshalType) {
+ this.unmarshalType = unmarshalType;
+ return this;
+ }
+
+ @Override
+ public ParquetAvroDataFormat end() {
+ return new ParquetAvroDataFormat(this);
+ }
+ }
+
+}
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/transformer/DataFormatTransformerDefinition.java b/core/camel-core-model/src/main/java/org/apache/camel/model/transformer/DataFormatTransformerDefinition.java
index c6fc23c13fb..d0fbe43f752 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/model/transformer/DataFormatTransformerDefinition.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/transformer/DataFormatTransformerDefinition.java
@@ -46,6 +46,7 @@ import org.apache.camel.model.dataformat.JsonDataFormat;
import org.apache.camel.model.dataformat.LZFDataFormat;
import org.apache.camel.model.dataformat.MimeMultipartDataFormat;
import org.apache.camel.model.dataformat.PGPDataFormat;
+import org.apache.camel.model.dataformat.ParquetAvroDataFormat;
import org.apache.camel.model.dataformat.ProtobufDataFormat;
import org.apache.camel.model.dataformat.RssDataFormat;
import org.apache.camel.model.dataformat.SoapDataFormat;
@@ -97,6 +98,7 @@ public class DataFormatTransformerDefinition extends TransformerDefinition {
@XmlElement(name = "jsonApi", type = JsonApiDataFormat.class),
@XmlElement(name = "lzf", type = LZFDataFormat.class),
@XmlElement(name = "mimeMultipart", type = MimeMultipartDataFormat.class),
+ @XmlElement(name = "parquetAvro", type = ParquetAvroDataFormat.class),
@XmlElement(name = "protobuf", type = ProtobufDataFormat.class),
@XmlElement(name = "rss", type = RssDataFormat.class),
@XmlElement(name = "soap", type = SoapDataFormat.class),
diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/dataformat/DataFormatReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/dataformat/DataFormatReifier.java
index 3e9a171f300..be920d3e058 100644
--- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/dataformat/DataFormatReifier.java
+++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/dataformat/DataFormatReifier.java
@@ -49,6 +49,7 @@ import org.apache.camel.model.dataformat.JsonDataFormat;
import org.apache.camel.model.dataformat.LZFDataFormat;
import org.apache.camel.model.dataformat.MimeMultipartDataFormat;
import org.apache.camel.model.dataformat.PGPDataFormat;
+import org.apache.camel.model.dataformat.ParquetAvroDataFormat;
import org.apache.camel.model.dataformat.ProtobufDataFormat;
import org.apache.camel.model.dataformat.RssDataFormat;
import org.apache.camel.model.dataformat.SoapDataFormat;
@@ -221,6 +222,8 @@ public abstract class DataFormatReifier<T extends DataFormatDefinition> extends
return new LZFDataFormatReifier(camelContext, definition);
} else if (definition instanceof MimeMultipartDataFormat) {
return new MimeMultipartDataFormatReifier(camelContext, definition);
+ } else if (definition instanceof ParquetAvroDataFormat) {
+ return new ParquetAvroDataFormatReifier(camelContext, definition);
} else if (definition instanceof PGPDataFormat) {
return new PGPDataFormatReifier(camelContext, definition);
} else if (definition instanceof ProtobufDataFormat) {
diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/dataformat/ParquetAvroDataFormatReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/dataformat/ParquetAvroDataFormatReifier.java
new file mode 100644
index 00000000000..eda9b12661e
--- /dev/null
+++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/dataformat/ParquetAvroDataFormatReifier.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.reifier.dataformat;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.model.DataFormatDefinition;
+import org.apache.camel.model.dataformat.ParquetAvroDataFormat;
+
+public class ParquetAvroDataFormatReifier extends DataFormatReifier<ParquetAvroDataFormat> {
+
+ public ParquetAvroDataFormatReifier(CamelContext camelContext, DataFormatDefinition definition) {
+ super(camelContext, (ParquetAvroDataFormat) definition);
+ }
+
+ @Override
+ protected void prepareDataFormatConfig(Map<String, Object> properties) {
+ properties.put("unmarshalType", or(definition.getUnmarshalType(), definition.getUnmarshalTypeName()));
+ }
+
+}
diff --git a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
index 01f6b773715..68a33cda6a7 100644
--- a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
+++ b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
@@ -2347,6 +2347,15 @@ public class ModelParser extends BaseParser {
return true;
}, noElementHandler(), noValueHandler());
}
+ protected ParquetAvroDataFormat doParseParquetAvroDataFormat() throws IOException, XmlPullParserException {
+ return doParse(new ParquetAvroDataFormat(), (def, key, val) -> {
+ if ("unmarshalType".equals(key)) {
+ def.setUnmarshalTypeName(val);
+ return true;
+ }
+ return identifiedTypeAttributeHandler().accept(def, key, val);
+ }, noElementHandler(), noValueHandler());
+ }
protected ProtobufDataFormat doParseProtobufDataFormat() throws IOException, XmlPullParserException {
return doParse(new ProtobufDataFormat(), (def, key, val) -> {
switch (key) {
@@ -3457,6 +3466,7 @@ public class ModelParser extends BaseParser {
case "lzf": return doParseLZFDataFormat();
case "mimeMultipart": return doParseMimeMultipartDataFormat();
case "pgp": return doParsePGPDataFormat();
+ case "parquetAvro": return doParseParquetAvroDataFormat();
case "protobuf": return doParseProtobufDataFormat();
case "rss": return doParseRssDataFormat();
case "soap": return doParseSoapDataFormat();
diff --git a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
index 0a2c5047d23..35a5d650473 100644
--- a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
+++ b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
@@ -685,6 +685,11 @@ public class ModelWriter extends BaseWriter {
public void writePGPDataFormat(PGPDataFormat def) throws IOException {
doWritePGPDataFormat("pgp", def);
}
+ public void writeParquetAvroDataFormat(
+ ParquetAvroDataFormat def)
+ throws IOException {
+ doWriteParquetAvroDataFormat("parquetAvro", def);
+ }
public void writeProtobufDataFormat(
ProtobufDataFormat def)
throws IOException {
@@ -1533,6 +1538,7 @@ public class ModelWriter extends BaseWriter {
case "JsonApiDataFormat" -> doWriteJsonApiDataFormat("jsonApi", (JsonApiDataFormat) def.getDataFormatType());
case "LZFDataFormat" -> doWriteLZFDataFormat("lzf", (LZFDataFormat) def.getDataFormatType());
case "MimeMultipartDataFormat" -> doWriteMimeMultipartDataFormat("mimeMultipart", (MimeMultipartDataFormat) def.getDataFormatType());
+ case "ParquetAvroDataFormat" -> doWriteParquetAvroDataFormat("parquetAvro", (ParquetAvroDataFormat) def.getDataFormatType());
case "ProtobufDataFormat" -> doWriteProtobufDataFormat("protobuf", (ProtobufDataFormat) def.getDataFormatType());
case "RssDataFormat" -> doWriteRssDataFormat("rss", (RssDataFormat) def.getDataFormatType());
case "SoapDataFormat" -> doWriteSoapDataFormat("soap", (SoapDataFormat) def.getDataFormatType());
@@ -2506,6 +2512,7 @@ public class ModelWriter extends BaseWriter {
case "JsonApiDataFormat" -> doWriteJsonApiDataFormat("jsonApi", (JsonApiDataFormat) def.getDataFormatType());
case "LZFDataFormat" -> doWriteLZFDataFormat("lzf", (LZFDataFormat) def.getDataFormatType());
case "MimeMultipartDataFormat" -> doWriteMimeMultipartDataFormat("mimeMultipart", (MimeMultipartDataFormat) def.getDataFormatType());
+ case "ParquetAvroDataFormat" -> doWriteParquetAvroDataFormat("parquetAvro", (ParquetAvroDataFormat) def.getDataFormatType());
case "ProtobufDataFormat" -> doWriteProtobufDataFormat("protobuf", (ProtobufDataFormat) def.getDataFormatType());
case "RssDataFormat" -> doWriteRssDataFormat("rss", (RssDataFormat) def.getDataFormatType());
case "SoapDataFormat" -> doWriteSoapDataFormat("soap", (SoapDataFormat) def.getDataFormatType());
@@ -3475,6 +3482,15 @@ public class ModelWriter extends BaseWriter {
doWriteAttribute("algorithm", def.getAlgorithm());
endElement();
}
+ protected void doWriteParquetAvroDataFormat(
+ String name,
+ ParquetAvroDataFormat def)
+ throws IOException {
+ startElement(name);
+ doWriteIdentifiedTypeAttributes(def);
+ doWriteAttribute("unmarshalType", def.getUnmarshalTypeName());
+ endElement();
+ }
protected void doWriteProtobufDataFormat(
String name,
ProtobufDataFormat def)
@@ -4592,6 +4608,7 @@ public class ModelWriter extends BaseWriter {
case "JsonApiDataFormat" -> doWriteJsonApiDataFormat("jsonApi", (JsonApiDataFormat) def.getDataFormatType());
case "LZFDataFormat" -> doWriteLZFDataFormat("lzf", (LZFDataFormat) def.getDataFormatType());
case "MimeMultipartDataFormat" -> doWriteMimeMultipartDataFormat("mimeMultipart", (MimeMultipartDataFormat) def.getDataFormatType());
+ case "ParquetAvroDataFormat" -> doWriteParquetAvroDataFormat("parquetAvro", (ParquetAvroDataFormat) def.getDataFormatType());
case "ProtobufDataFormat" -> doWriteProtobufDataFormat("protobuf", (ProtobufDataFormat) def.getDataFormatType());
case "RssDataFormat" -> doWriteRssDataFormat("rss", (RssDataFormat) def.getDataFormatType());
case "SoapDataFormat" -> doWriteSoapDataFormat("soap", (SoapDataFormat) def.getDataFormatType());
diff --git a/dsl/camel-endpointdsl/pom.xml b/dsl/camel-endpointdsl/pom.xml
index fe8a504e9c1..b641fffd4c5 100644
--- a/dsl/camel-endpointdsl/pom.xml
+++ b/dsl/camel-endpointdsl/pom.xml
@@ -44,6 +44,11 @@
<groupId>org.apache.camel</groupId>
<artifactId>camel-core-engine</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.sun.mail</groupId>
+ <artifactId>jakarta.mail</artifactId>
+ <version>1.6.7</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-allcomponents</artifactId>
diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java b/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
index c21941957bd..f12137d665b 100644
--- a/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
+++ b/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
@@ -151,6 +151,7 @@ import org.apache.camel.model.dataformat.JsonDataFormat;
import org.apache.camel.model.dataformat.LZFDataFormat;
import org.apache.camel.model.dataformat.MimeMultipartDataFormat;
import org.apache.camel.model.dataformat.PGPDataFormat;
+import org.apache.camel.model.dataformat.ParquetAvroDataFormat;
import org.apache.camel.model.dataformat.ProtobufDataFormat;
import org.apache.camel.model.dataformat.RssDataFormat;
import org.apache.camel.model.dataformat.SoapDataFormat;
@@ -3144,6 +3145,7 @@ public final class ModelDeserializers extends YamlDeserializerSupport {
@YamlProperty(name = "json-api", type = "object:org.apache.camel.model.dataformat.JsonApiDataFormat"),
@YamlProperty(name = "lzf", type = "object:org.apache.camel.model.dataformat.LZFDataFormat"),
@YamlProperty(name = "mime-multipart", type = "object:org.apache.camel.model.dataformat.MimeMultipartDataFormat"),
+ @YamlProperty(name = "parquet-avro", type = "object:org.apache.camel.model.dataformat.ParquetAvroDataFormat"),
@YamlProperty(name = "pgp", type = "object:org.apache.camel.model.dataformat.PGPDataFormat"),
@YamlProperty(name = "protobuf", type = "object:org.apache.camel.model.dataformat.ProtobufDataFormat"),
@YamlProperty(name = "rss", type = "object:org.apache.camel.model.dataformat.RssDataFormat"),
@@ -3294,6 +3296,11 @@ public final class ModelDeserializers extends YamlDeserializerSupport {
target.setDataFormatType(val);
break;
}
+ case "parquet-avro": {
+ org.apache.camel.model.dataformat.ParquetAvroDataFormat val = asType(node, org.apache.camel.model.dataformat.ParquetAvroDataFormat.class);
+ target.setDataFormatType(val);
+ break;
+ }
case "protobuf": {
org.apache.camel.model.dataformat.ProtobufDataFormat val = asType(node, org.apache.camel.model.dataformat.ProtobufDataFormat.class);
target.setDataFormatType(val);
@@ -8417,6 +8424,7 @@ public final class ModelDeserializers extends YamlDeserializerSupport {
@YamlProperty(name = "json-api", type = "object:org.apache.camel.model.dataformat.JsonApiDataFormat"),
@YamlProperty(name = "lzf", type = "object:org.apache.camel.model.dataformat.LZFDataFormat"),
@YamlProperty(name = "mime-multipart", type = "object:org.apache.camel.model.dataformat.MimeMultipartDataFormat"),
+ @YamlProperty(name = "parquet-avro", type = "object:org.apache.camel.model.dataformat.ParquetAvroDataFormat"),
@YamlProperty(name = "pgp", type = "object:org.apache.camel.model.dataformat.PGPDataFormat"),
@YamlProperty(name = "protobuf", type = "object:org.apache.camel.model.dataformat.ProtobufDataFormat"),
@YamlProperty(name = "rss", type = "object:org.apache.camel.model.dataformat.RssDataFormat"),
@@ -8565,6 +8573,11 @@ public final class ModelDeserializers extends YamlDeserializerSupport {
target.setDataFormatType(val);
break;
}
+ case "parquet-avro": {
+ org.apache.camel.model.dataformat.ParquetAvroDataFormat val = asType(node, org.apache.camel.model.dataformat.ParquetAvroDataFormat.class);
+ target.setDataFormatType(val);
+ break;
+ }
case "protobuf": {
org.apache.camel.model.dataformat.ProtobufDataFormat val = asType(node, org.apache.camel.model.dataformat.ProtobufDataFormat.class);
target.setDataFormatType(val);
@@ -10086,6 +10099,56 @@ public final class ModelDeserializers extends YamlDeserializerSupport {
}
}
+ @YamlType(
+ nodes = {
+ "parquet-avro",
+ "parquetAvro"
+ },
+ inline = true,
+ types = org.apache.camel.model.dataformat.ParquetAvroDataFormat.class,
+ order = org.apache.camel.dsl.yaml.common.YamlDeserializerResolver.ORDER_LOWEST - 1,
+ properties = {
+ @YamlProperty(name = "id", type = "string"),
+ @YamlProperty(name = "unmarshal-type", type = "string")
+ }
+ )
+ public static class ParquetAvroDataFormatDeserializer extends YamlDeserializerBase<ParquetAvroDataFormat> {
+ public ParquetAvroDataFormatDeserializer() {
+ super(ParquetAvroDataFormat.class);
+ }
+
+ @Override
+ protected ParquetAvroDataFormat newInstance() {
+ return new ParquetAvroDataFormat();
+ }
+
+ @Override
+ protected ParquetAvroDataFormat newInstance(String value) {
+ return new ParquetAvroDataFormat(value);
+ }
+
+ @Override
+ protected boolean setProperty(ParquetAvroDataFormat target, String propertyKey,
+ String propertyName, Node node) {
+ switch(propertyKey) {
+ case "id": {
+ String val = asText(node);
+ target.setId(val);
+ break;
+ }
+ case "unmarshal-type": {
+ String val = asText(node);
+ target.setUnmarshalTypeName(val);
+ break;
+ }
+ default: {
+ return false;
+ }
+ }
+ return true;
+ }
+ }
+
@YamlType(
nodes = {
"pass-through-service-filter",
@@ -17704,6 +17767,7 @@ public final class ModelDeserializers extends YamlDeserializerSupport {
@YamlProperty(name = "json-api", type = "object:org.apache.camel.model.dataformat.JsonApiDataFormat"),
@YamlProperty(name = "lzf", type = "object:org.apache.camel.model.dataformat.LZFDataFormat"),
@YamlProperty(name = "mime-multipart", type = "object:org.apache.camel.model.dataformat.MimeMultipartDataFormat"),
+ @YamlProperty(name = "parquet-avro", type = "object:org.apache.camel.model.dataformat.ParquetAvroDataFormat"),
@YamlProperty(name = "pgp", type = "object:org.apache.camel.model.dataformat.PGPDataFormat"),
@YamlProperty(name = "protobuf", type = "object:org.apache.camel.model.dataformat.ProtobufDataFormat"),
@YamlProperty(name = "rss", type = "object:org.apache.camel.model.dataformat.RssDataFormat"),
@@ -17857,6 +17921,11 @@ public final class ModelDeserializers extends YamlDeserializerSupport {
target.setDataFormatType(val);
break;
}
+ case "parquet-avro": {
+ org.apache.camel.model.dataformat.ParquetAvroDataFormat val = asType(node, org.apache.camel.model.dataformat.ParquetAvroDataFormat.class);
+ target.setDataFormatType(val);
+ break;
+ }
case "protobuf": {
org.apache.camel.model.dataformat.ProtobufDataFormat val = asType(node, org.apache.camel.model.dataformat.ProtobufDataFormat.class);
target.setDataFormatType(val);
diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializersResolver.java b/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializersResolver.java
index 0165a5f5ffe..3e2b9a8516a 100644
--- a/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializersResolver.java
+++ b/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializersResolver.java
@@ -281,6 +281,9 @@ public final class ModelDeserializersResolver implements YamlDeserializerResolve
case "org.apache.camel.model.PackageScanDefinition": return new ModelDeserializers.PackageScanDefinitionDeserializer();
case "param": return new ModelDeserializers.ParamDefinitionDeserializer();
case "org.apache.camel.model.rest.ParamDefinition": return new ModelDeserializers.ParamDefinitionDeserializer();
+ case "parquet-avro": return new ModelDeserializers.ParquetAvroDataFormatDeserializer();
+ case "parquetAvro": return new ModelDeserializers.ParquetAvroDataFormatDeserializer();
+ case "org.apache.camel.model.dataformat.ParquetAvroDataFormat": return new ModelDeserializers.ParquetAvroDataFormatDeserializer();
case "pass-through-service-filter": return new ModelDeserializers.PassThroughServiceCallServiceFilterConfigurationDeserializer();
case "passThroughServiceFilter": return new ModelDeserializers.PassThroughServiceCallServiceFilterConfigurationDeserializer();
case "org.apache.camel.model.cloud.PassThroughServiceCallServiceFilterConfiguration": return new ModelDeserializers.PassThroughServiceCallServiceFilterConfigurationDeserializer();
diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camel-yaml-dsl.json b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camel-yaml-dsl.json
index 569f134c960..497be080d40 100644
--- a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camel-yaml-dsl.json
+++ b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camel-yaml-dsl.json
@@ -1454,6 +1454,9 @@
"mime-multipart" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.MimeMultipartDataFormat"
},
+ "parquet-avro" : {
+ "$ref" : "#/items/definitions/org.apache.camel.model.dataformat.ParquetAvroDataFormat"
+ },
"pgp" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.PGPDataFormat"
},
@@ -3449,6 +3452,9 @@
"mime-multipart" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.MimeMultipartDataFormat"
},
+ "parquet-avro" : {
+ "$ref" : "#/items/definitions/org.apache.camel.model.dataformat.ParquetAvroDataFormat"
+ },
"pgp" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.PGPDataFormat"
},
@@ -5421,6 +5427,21 @@
}
}
},
+ "org.apache.camel.model.dataformat.ParquetAvroDataFormat" : {
+ "oneOf" : [ {
+ "type" : "string"
+ }, {
+ "type" : "object",
+ "properties" : {
+ "id" : {
+ "type" : "string"
+ },
+ "unmarshal-type" : {
+ "type" : "string"
+ }
+ }
+ } ]
+ },
"org.apache.camel.model.dataformat.ProtobufDataFormat" : {
"oneOf" : [ {
"type" : "string"
@@ -8054,6 +8075,9 @@
"mime-multipart" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.MimeMultipartDataFormat"
},
+ "parquet-avro" : {
+ "$ref" : "#/items/definitions/org.apache.camel.model.dataformat.ParquetAvroDataFormat"
+ },
"pgp" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.PGPDataFormat"
},
diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
index 79fcf7c1a8d..dcfc72d2984 100644
--- a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
+++ b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
@@ -1364,6 +1364,9 @@
"mimeMultipart" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.MimeMultipartDataFormat"
},
+ "parquetAvro" : {
+ "$ref" : "#/items/definitions/org.apache.camel.model.dataformat.ParquetAvroDataFormat"
+ },
"pgp" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.PGPDataFormat"
},
@@ -3359,6 +3362,9 @@
"mimeMultipart" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.MimeMultipartDataFormat"
},
+ "parquetAvro" : {
+ "$ref" : "#/items/definitions/org.apache.camel.model.dataformat.ParquetAvroDataFormat"
+ },
"pgp" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.PGPDataFormat"
},
@@ -5331,6 +5337,21 @@
}
}
},
+ "org.apache.camel.model.dataformat.ParquetAvroDataFormat" : {
+ "oneOf" : [ {
+ "type" : "string"
+ }, {
+ "type" : "object",
+ "properties" : {
+ "id" : {
+ "type" : "string"
+ },
+ "unmarshalType" : {
+ "type" : "string"
+ }
+ }
+ } ]
+ },
"org.apache.camel.model.dataformat.ProtobufDataFormat" : {
"oneOf" : [ {
"type" : "string"
@@ -7961,6 +7982,9 @@
"mimeMultipart" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.MimeMultipartDataFormat"
},
+ "parquetAvro" : {
+ "$ref" : "#/items/definitions/org.apache.camel.model.dataformat.ParquetAvroDataFormat"
+ },
"pgp" : {
"$ref" : "#/items/definitions/org.apache.camel.model.dataformat.PGPDataFormat"
},
diff --git a/parent/pom.xml b/parent/pom.xml
index d6794edafbf..8369be92079 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -403,6 +403,8 @@
<optaplanner-version>9.37.0.Final</optaplanner-version>
<os-maven-plugin-version>1.7.0</os-maven-plugin-version>
<paho-version>1.2.5</paho-version>
+ <parquet-common-version>1.12.0</parquet-common-version>
+ <parquet-avro-version>1.12.3</parquet-avro-version>
<pdfbox-version>2.0.28</pdfbox-version>
<pgjdbc-driver-version>42.6.0</pgjdbc-driver-version>
<pgjdbc-ng-driver-version>0.8.9</pgjdbc-ng-driver-version>
@@ -2455,6 +2457,11 @@
<artifactId>dummy-component</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel.dataformat.parquet.avro</groupId>
+ <artifactId>camel-parquet-avro</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel.maven</groupId>
<artifactId>camel-debezium-maven-plugin</artifactId>