You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by shubham-pathak22 <gi...@git.apache.org> on 2016/03/22 11:24:18 UTC
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
GitHub user shubham-pathak22 opened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/219
APEXMALHAR-2014 added parquet reader
Adding **ParquetReaderOperator** which will allow apex users to read parquet files.
**Apache Parquet** is a columnar storage format available to any project in the Hadoop ecosystem, regardless of the choice of data processing framework, data model or programming language.
For more information : [Apache Parquet] (https://parquet.apache.org/documentation/latest/ "Apache Parquet")
#### Implementation Details
* **AbstractParquetFileReaderOperator** extends from **AbstractFileInputOperator**. Overrides *openFile()* and *readEntity()* methods.
* *openFile()* method instantiates a *ParquetReader* ( reader provided by parquet-mr project that reads parquet records from a file ) with *GroupReadSupport* ( records would be read as *Group* ) .
* *readEntity()* method reads the records and calls *convertGroup()* method. Derived classes to override *convertGroup()* method to convert *Group* to any form required by downstream operators.
* Provided **ParquetFilePOJOReader** operator which is a concrete implementation of **AbstractParquetFileReader** to read Parquet files and emits records as POJOs. The
POJO class name & field mapping should be provided by the user. If this
mapping is not provided then reflection is used to determine this mapping. As
of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
are supported.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/shubham-pathak22/incubator-apex-malhar APEXMALHAR-2014
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/219.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #219
----
commit d5c9415959f583caf68292cfb3cfac262d49eb99
Author: shubham <sh...@github.com>
Date: 2016-03-21T10:20:21Z
APEXMALHAR-2014 added parquet reader
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58043245
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/AbstractParquetFileReader.java ---
@@ -0,0 +1,113 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.AbstractFileInputOperator;
+
+import parquet.example.data.Group;
+import parquet.hadoop.ParquetReader;
+import parquet.hadoop.example.GroupReadSupport;
+import parquet.schema.MessageType;
+import parquet.schema.MessageTypeParser;
+
+/**
+ * Base implementation of ParquetFileReader. Reads parquet files from input
+ * directory using GroupReadSupport. Derived classes need to implement
+ * {@link #convertGroup(Group)} method to convert Group to other type. Example
+ * of such implementation is {@link ParquetFilePOJOReader} that converts Group
+ * to POJO.
+ *
+ * @since 3.3.3
+ */
+public abstract class AbstractParquetFileReader<T> extends AbstractFileInputOperator<T>
+{
+ private transient ParquetReader<Group> reader;
+ protected transient MessageType schema;
+ protected String parquetSchema;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ schema = MessageTypeParser.parseMessageType(parquetSchema);
+ }
+
+ /**
+ * Opens the file to read using GroupReadSupport
+ */
+ @Override
+ protected InputStream openFile(Path path) throws IOException
+ {
+ InputStream is = super.openFile(path);
+ GroupReadSupport readSupport = new GroupReadSupport();
+ readSupport.init(configuration, null, schema);
+ reader = new ParquetReader<Group>(path, readSupport);
+ return is;
+ }
+
+ /**
+ * Reads next record in parquet file as a group. Returns null when end of file
+ * is reached
+ */
+ @Override
+ protected T readEntity() throws IOException
+ {
+ Group group = reader.read();
+ if (group != null) {
+ return convertGroup(group);
+ }
+ return null;
+ }
+
+ /**
+ * Converts Group to object of type T. Derived classes need to provide
+ * implementation for conversion
+ *
+ * @param Group
+ * @return object of type T
+ */
+ protected abstract T convertGroup(Group group);
+
+ /**
+ * Get Parquet Schema
+ *
+ * @return parquetSchema
--- End diff --
Description missing for parquetSchema.. Same in setter method.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58044255
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/AbstractParquetFileReader.java ---
@@ -0,0 +1,113 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.AbstractFileInputOperator;
+
+import parquet.example.data.Group;
+import parquet.hadoop.ParquetReader;
+import parquet.hadoop.example.GroupReadSupport;
+import parquet.schema.MessageType;
+import parquet.schema.MessageTypeParser;
+
+/**
+ * Base implementation of ParquetFileReader. Reads parquet files from input
+ * directory using GroupReadSupport. Derived classes need to implement
+ * {@link #convertGroup(Group)} method to convert Group to other type. Example
+ * of such implementation is {@link ParquetFilePOJOReader} that converts Group
+ * to POJO.
+ *
+ * @since 3.3.3
+ */
+public abstract class AbstractParquetFileReader<T> extends AbstractFileInputOperator<T>
--- End diff --
Are there places other than ParquetFilePOJOReader.java where this this abstract class can be extended... If not can we merge these 2 files. Just want to avoid management of multiple files.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58046072
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.InvalidRecordException;
+import parquet.io.ParquetEncodingException;
+import parquet.schema.PrimitiveType.PrimitiveTypeName;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * Parquet primitive types maps to POJO datatypes as follows : INT32 -> int
+ * INT64 -> long BOOLEAN -> boolean FLOAT -> float DOUBLE -> double BINARY ->
+ * String
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
+ * POJO class
+ */
+ protected transient Class<?> pojoClass;
+ /**
+ * Map containing setters for fields in POJO
+ */
+ protected transient Map<String, Setter> pojoSetters;
+ /**
+ * String representing Parquet TO POJO field mapping. If not provided, then
+ * reflection is used to determine the mapping. Format :
+ * PARQUET_FIELD_NAME:POJO_FIELD_NAME:TYPE
+ * E.g.event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:
+ * long_id_v2:
+ * LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:
+ * FLOAT,double_val:double_val_v2:DOUBLE
+ */
+ protected transient String groupToPOJOFieldsMapping = null;
+ protected transient List<FieldInfo> fieldInfos;
+ protected transient List<ActiveFieldInfo> columnFieldSetters = null;
+ protected static final String FIELD_SEPARATOR = ":";
+ protected static final String RECORD_SEPARATOR = ",";
+ private static final Logger logger = LoggerFactory.getLogger(ParquetFilePOJOReader.class);
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
+ {
+
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ pojoClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ pojoSetters = Maps.newHashMap();
+ for (Field f : pojoClass.getDeclaredFields()) {
+ try {
+ pojoSetters.put(f.getName(), generateSettersForField(pojoClass, f.getName()));
+ } catch (NoSuchFieldException | SecurityException e) {
+ logger.error("Failed to generate setters. Exception {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ if (groupToPOJOFieldsMapping == null) {
+ fieldInfos = createFieldInfoMap(generateFieldInfoInputs());
+ } else {
+ fieldInfos = createFieldInfoMap(groupToPOJOFieldsMapping);
+ }
+ initColumnFieldSetters();
+ }
+
+ };
+
+ /**
+ * Converts Group to POJO
+ */
+ @Override
+ protected Object convertGroup(Group group)
+ {
+ Object obj;
+ try {
+ obj = pojoClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException ex) {
+ throw new RuntimeException(ex);
+ }
+ for (int i = 0; i < columnFieldSetters.size(); i++) {
+ try {
+ ParquetFilePOJOReader.ActiveFieldInfo afi = columnFieldSetters.get(i);
+ int fieldIndex = schema.getFieldIndex(afi.fieldInfo.getColumnName());
--- End diff --
fieldIndex can be useful for instead of setter map.. Maps are costly to retrieve data from.
If you could have an array where index is fieldIndex and value at that index is corresponding setter method, the access would be much faster. This would improve the processing speed of a tuple.
For eg. PojoUtils.Setters[] pojoSetters;
and while accessing you could do:
pojoSetters[fieldIndex].set(obj, value);
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by devtagare <gi...@git.apache.org>.
Github user devtagare commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r57661028
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,305 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.ParquetEncodingException;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
--- End diff --
Please add a Logger and use it in required places
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58043621
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.InvalidRecordException;
+import parquet.io.ParquetEncodingException;
+import parquet.schema.PrimitiveType.PrimitiveTypeName;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * Parquet primitive types maps to POJO datatypes as follows : INT32 -> int
+ * INT64 -> long BOOLEAN -> boolean FLOAT -> float DOUBLE -> double BINARY ->
+ * String
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
+ * POJO class
+ */
+ protected transient Class<?> pojoClass;
+ /**
+ * Map containing setters for fields in POJO
+ */
+ protected transient Map<String, Setter> pojoSetters;
+ /**
+ * String representing Parquet TO POJO field mapping. If not provided, then
+ * reflection is used to determine the mapping. Format :
+ * PARQUET_FIELD_NAME:POJO_FIELD_NAME:TYPE
+ * E.g.event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:
+ * long_id_v2:
+ * LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:
+ * FLOAT,double_val:double_val_v2:DOUBLE
+ */
+ protected transient String groupToPOJOFieldsMapping = null;
+ protected transient List<FieldInfo> fieldInfos;
+ protected transient List<ActiveFieldInfo> columnFieldSetters = null;
+ protected static final String FIELD_SEPARATOR = ":";
+ protected static final String RECORD_SEPARATOR = ",";
+ private static final Logger logger = LoggerFactory.getLogger(ParquetFilePOJOReader.class);
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
+ {
+
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ pojoClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ pojoSetters = Maps.newHashMap();
+ for (Field f : pojoClass.getDeclaredFields()) {
+ try {
+ pojoSetters.put(f.getName(), generateSettersForField(pojoClass, f.getName()));
+ } catch (NoSuchFieldException | SecurityException e) {
+ logger.error("Failed to generate setters. Exception {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ if (groupToPOJOFieldsMapping == null) {
+ fieldInfos = createFieldInfoMap(generateFieldInfoInputs());
+ } else {
+ fieldInfos = createFieldInfoMap(groupToPOJOFieldsMapping);
+ }
+ initColumnFieldSetters();
+ }
+
+ };
+
+ /**
+ * Converts Group to POJO
+ */
+ @Override
+ protected Object convertGroup(Group group)
+ {
+ Object obj;
+ try {
+ obj = pojoClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException ex) {
+ throw new RuntimeException(ex);
+ }
+ for (int i = 0; i < columnFieldSetters.size(); i++) {
+ try {
+ ParquetFilePOJOReader.ActiveFieldInfo afi = columnFieldSetters.get(i);
+ int fieldIndex = schema.getFieldIndex(afi.fieldInfo.getColumnName());
+ PrimitiveTypeName primitiveTypeName = schema.getType(fieldIndex).asPrimitiveType().getPrimitiveTypeName();
+
+ switch (primitiveTypeName) {
+
+ case BOOLEAN:
+ Boolean booleanVal = Boolean.parseBoolean(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, booleanVal);
+ break;
+
+ case INT32:
+ Integer intVal = Integer.parseInt(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, intVal);
+ break;
+
+ case INT64:
+ Long longVal = Long.parseLong(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, longVal);
+ break;
+
+ case FLOAT:
+ Float floatVal = Float.parseFloat(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, floatVal);
+ break;
+
+ case DOUBLE:
+ Double doubleVal = Double.parseDouble(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, doubleVal);
+ break;
+
+ case BINARY:
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, group.getValueToString(fieldIndex, 0));
+ break;
+
+ default:
+ throw new ParquetEncodingException("Unsupported column type: " + primitiveTypeName);
+
+ }
+ } catch (InvalidRecordException e) {
+ logger.error("Field not found in schema {} ", e);
+ }
+ }
+ return obj;
+ }
+
+ @Override
+ protected void emit(Object tuple)
+ {
+ output.emit(tuple);
+ }
+
+ /**
+ * Sets the {@link FieldInfo}s. A {@link FieldInfo} maps a store column to a
+ * pojo field name.<br/>
+ * The value from fieldInfo.column is assigned to
+ * fieldInfo.pojoFieldExpression.
+ *
+ * @description $[].columnName name of the Output Field in POJO
+ * @description $[].pojoFieldExpression expression to get the respective field
+ * from generic record
+ * @useSchema $[].pojoFieldExpression outputPort.fields[].name
+ */
+ public void setFieldInfos(List<FieldInfo> fieldInfos)
+ {
+ this.fieldInfos = fieldInfos;
+ }
+
+ /**
+ * Creates a map representing fieldName in POJO:field in Generic Record:Data
+ * type
+ *
+ * @return List of FieldInfo
+ */
+ public List<FieldInfo> createFieldInfoMap(String str)
+ {
+ fieldInfos = new ArrayList<FieldInfo>();
+
+ StringTokenizer strtok = new StringTokenizer(str, RECORD_SEPARATOR);
+
+ while (strtok.hasMoreTokens()) {
+ String[] token = strtok.nextToken().split(FIELD_SEPARATOR);
+
+ fieldInfos.add(new FieldInfo(token[0], token[1], SupportType.valueOf(token[2])));
+ }
+
+ return fieldInfos;
+ }
+
+ /**
+ * Generates setter for given field and POJO class
+ *
+ * @param klass
--- End diff --
Please add descriptions for params in javadoc.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by devtagare <gi...@git.apache.org>.
Github user devtagare commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r57661266
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,305 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.ParquetEncodingException;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
+ * POJO class
+ */
+ protected transient Class<?> pojoClass;
+ /**
+ * Map containing setters for fields in POJO
+ */
+ protected transient Map<String, Setter> pojoSetters;
+ /**
+ * String representing Parquet TO POJO field mapping. If not provided, then
+ * reflection is used to determine the mapping. Format :
+ * PARQUET_FIELD_NAME:POJO_FIELD_NAME:TYPE
+ * E.g.event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:
+ * long_id_v2:
+ * LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:
+ * FLOAT,double_val:double_val_v2:DOUBLE
+ */
+ protected transient String groupToPOJOFieldsMapping = null;
+ protected transient List<FieldInfo> fieldInfos;
+ protected transient List<ActiveFieldInfo> columnFieldSetters = null;
+ protected static final String FIELD_SEPARATOR = ":";
+ protected static final String RECORD_SEPARATOR = ",";
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
+ {
+
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ pojoClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ pojoSetters = Maps.newHashMap();
+ for (Field f : pojoClass.getDeclaredFields()) {
+ try {
+ pojoSetters.put(f.getName(), generateSettersForField(pojoClass, f.getName()));
+ } catch (NoSuchFieldException | SecurityException e) {
+ e.printStackTrace();
+ }
+ }
+
+ if (groupToPOJOFieldsMapping == null) {
+ fieldInfos = createFieldInfoMap(generateFieldInfoInputs());
+ } else {
+ fieldInfos = createFieldInfoMap(groupToPOJOFieldsMapping);
+ }
+ initColumnFieldSetters();
+ }
+
+ };
+
+ /**
+ * Converts Group to POJO
+ */
+ @Override
+ protected Object convertGroup(Group group)
+ {
+ Object obj;
+ try {
+ obj = pojoClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException ex) {
+ throw new RuntimeException(ex);
+ }
+ for (int i = 0; i < columnFieldSetters.size(); i++) {
+ ParquetFilePOJOReader.ActiveFieldInfo afi = columnFieldSetters.get(i);
+
+ afi = columnFieldSetters.get(i);
+ int fieldIndex = schema.getFieldIndex(afi.fieldInfo.getColumnName());
+ SupportType st = afi.fieldInfo.getType();
+
+ switch (st) {
+
+ case BOOLEAN:
+ Boolean booleanVal = Boolean.parseBoolean(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, booleanVal);
+ break;
+
+ case INTEGER:
+ Integer intVal = Integer.parseInt(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, intVal);
+ break;
+
+ case LONG:
+ Long longVal = Long.parseLong(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, longVal);
+ break;
+
+ case FLOAT:
+ Float floatVal = Float.parseFloat(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, floatVal);
+ break;
+
+ case DOUBLE:
+ Double doubleVal = Double.parseDouble(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, doubleVal);
+ break;
+
+ case STRING:
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, group.getValueToString(fieldIndex, 0));
+ break;
+
+ default:
+ throw new ParquetEncodingException("Unsupported column type: " + schema.getType(fieldIndex));
--- End diff --
Can the default case be to serialize the record as a byte array and set it on the object ? This could help serializing non-primitive types as byte arrays, which the user can then deserialize.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58044317
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.InvalidRecordException;
+import parquet.io.ParquetEncodingException;
+import parquet.schema.PrimitiveType.PrimitiveTypeName;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * Parquet primitive types maps to POJO datatypes as follows : INT32 -> int
+ * INT64 -> long BOOLEAN -> boolean FLOAT -> float DOUBLE -> double BINARY ->
+ * String
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
--- End diff --
Please remove @since
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by shubham-pathak22 <gi...@git.apache.org>.
Github user shubham-pathak22 closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/219
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58045610
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.InvalidRecordException;
+import parquet.io.ParquetEncodingException;
+import parquet.schema.PrimitiveType.PrimitiveTypeName;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * Parquet primitive types maps to POJO datatypes as follows : INT32 -> int
+ * INT64 -> long BOOLEAN -> boolean FLOAT -> float DOUBLE -> double BINARY ->
+ * String
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
+ * POJO class
+ */
+ protected transient Class<?> pojoClass;
+ /**
+ * Map containing setters for fields in POJO
+ */
+ protected transient Map<String, Setter> pojoSetters;
+ /**
+ * String representing Parquet TO POJO field mapping. If not provided, then
+ * reflection is used to determine the mapping. Format :
+ * PARQUET_FIELD_NAME:POJO_FIELD_NAME:TYPE
+ * E.g.event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:
+ * long_id_v2:
+ * LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:
+ * FLOAT,double_val:double_val_v2:DOUBLE
+ */
+ protected transient String groupToPOJOFieldsMapping = null;
+ protected transient List<FieldInfo> fieldInfos;
+ protected transient List<ActiveFieldInfo> columnFieldSetters = null;
+ protected static final String FIELD_SEPARATOR = ":";
+ protected static final String RECORD_SEPARATOR = ",";
+ private static final Logger logger = LoggerFactory.getLogger(ParquetFilePOJOReader.class);
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
+ {
+
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ pojoClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ pojoSetters = Maps.newHashMap();
+ for (Field f : pojoClass.getDeclaredFields()) {
+ try {
+ pojoSetters.put(f.getName(), generateSettersForField(pojoClass, f.getName()));
+ } catch (NoSuchFieldException | SecurityException e) {
+ logger.error("Failed to generate setters. Exception {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ if (groupToPOJOFieldsMapping == null) {
+ fieldInfos = createFieldInfoMap(generateFieldInfoInputs());
+ } else {
+ fieldInfos = createFieldInfoMap(groupToPOJOFieldsMapping);
+ }
+ initColumnFieldSetters();
+ }
+
+ };
+
+ /**
+ * Converts Group to POJO
+ */
+ @Override
+ protected Object convertGroup(Group group)
+ {
+ Object obj;
+ try {
+ obj = pojoClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException ex) {
+ throw new RuntimeException(ex);
+ }
+ for (int i = 0; i < columnFieldSetters.size(); i++) {
+ try {
+ ParquetFilePOJOReader.ActiveFieldInfo afi = columnFieldSetters.get(i);
+ int fieldIndex = schema.getFieldIndex(afi.fieldInfo.getColumnName());
+ PrimitiveTypeName primitiveTypeName = schema.getType(fieldIndex).asPrimitiveType().getPrimitiveTypeName();
+
+ switch (primitiveTypeName) {
+
+ case BOOLEAN:
+ Boolean booleanVal = Boolean.parseBoolean(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, booleanVal);
+ break;
+
+ case INT32:
+ Integer intVal = Integer.parseInt(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, intVal);
+ break;
+
+ case INT64:
+ Long longVal = Long.parseLong(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, longVal);
+ break;
+
+ case FLOAT:
+ Float floatVal = Float.parseFloat(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, floatVal);
+ break;
+
+ case DOUBLE:
+ Double doubleVal = Double.parseDouble(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, doubleVal);
+ break;
+
+ case BINARY:
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, group.getValueToString(fieldIndex, 0));
--- End diff --
Can we set this to byte[] instead... Setting toString may not be best thing to do here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58043536
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.InvalidRecordException;
+import parquet.io.ParquetEncodingException;
+import parquet.schema.PrimitiveType.PrimitiveTypeName;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * Parquet primitive types maps to POJO datatypes as follows : INT32 -> int
+ * INT64 -> long BOOLEAN -> boolean FLOAT -> float DOUBLE -> double BINARY ->
+ * String
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
+ * POJO class
+ */
+ protected transient Class<?> pojoClass;
+ /**
+ * Map containing setters for fields in POJO
+ */
+ protected transient Map<String, Setter> pojoSetters;
+ /**
+ * String representing Parquet TO POJO field mapping. If not provided, then
+ * reflection is used to determine the mapping. Format :
+ * PARQUET_FIELD_NAME:POJO_FIELD_NAME:TYPE
+ * E.g.event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:
+ * long_id_v2:
+ * LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:
+ * FLOAT,double_val:double_val_v2:DOUBLE
+ */
+ protected transient String groupToPOJOFieldsMapping = null;
+ protected transient List<FieldInfo> fieldInfos;
+ protected transient List<ActiveFieldInfo> columnFieldSetters = null;
+ protected static final String FIELD_SEPARATOR = ":";
+ protected static final String RECORD_SEPARATOR = ",";
+ private static final Logger logger = LoggerFactory.getLogger(ParquetFilePOJOReader.class);
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
+ {
+
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ pojoClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ pojoSetters = Maps.newHashMap();
+ for (Field f : pojoClass.getDeclaredFields()) {
+ try {
+ pojoSetters.put(f.getName(), generateSettersForField(pojoClass, f.getName()));
+ } catch (NoSuchFieldException | SecurityException e) {
+ logger.error("Failed to generate setters. Exception {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ if (groupToPOJOFieldsMapping == null) {
+ fieldInfos = createFieldInfoMap(generateFieldInfoInputs());
+ } else {
+ fieldInfos = createFieldInfoMap(groupToPOJOFieldsMapping);
+ }
+ initColumnFieldSetters();
+ }
+
+ };
+
+ /**
+ * Converts Group to POJO
+ */
+ @Override
+ protected Object convertGroup(Group group)
+ {
+ Object obj;
+ try {
+ obj = pojoClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException ex) {
+ throw new RuntimeException(ex);
+ }
+ for (int i = 0; i < columnFieldSetters.size(); i++) {
+ try {
+ ParquetFilePOJOReader.ActiveFieldInfo afi = columnFieldSetters.get(i);
+ int fieldIndex = schema.getFieldIndex(afi.fieldInfo.getColumnName());
+ PrimitiveTypeName primitiveTypeName = schema.getType(fieldIndex).asPrimitiveType().getPrimitiveTypeName();
+
+ switch (primitiveTypeName) {
+
+ case BOOLEAN:
+ Boolean booleanVal = Boolean.parseBoolean(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, booleanVal);
+ break;
+
+ case INT32:
+ Integer intVal = Integer.parseInt(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, intVal);
+ break;
+
+ case INT64:
+ Long longVal = Long.parseLong(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, longVal);
+ break;
+
+ case FLOAT:
+ Float floatVal = Float.parseFloat(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, floatVal);
+ break;
+
+ case DOUBLE:
+ Double doubleVal = Double.parseDouble(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, doubleVal);
+ break;
+
+ case BINARY:
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, group.getValueToString(fieldIndex, 0));
+ break;
+
+ default:
+ throw new ParquetEncodingException("Unsupported column type: " + primitiveTypeName);
+
+ }
+ } catch (InvalidRecordException e) {
+ logger.error("Field not found in schema {} ", e);
+ }
+ }
+ return obj;
+ }
+
+ @Override
+ protected void emit(Object tuple)
+ {
+ output.emit(tuple);
+ }
+
+ /**
+ * Sets the {@link FieldInfo}s. A {@link FieldInfo} maps a store column to a
+ * pojo field name.<br/>
+ * The value from fieldInfo.column is assigned to
+ * fieldInfo.pojoFieldExpression.
+ *
+ * @description $[].columnName name of the Output Field in POJO
+ * @description $[].pojoFieldExpression expression to get the respective field
+ * from generic record
+ * @useSchema $[].pojoFieldExpression outputPort.fields[].name
+ */
+ public void setFieldInfos(List<FieldInfo> fieldInfos)
+ {
+ this.fieldInfos = fieldInfos;
+ }
+
+ /**
+ * Creates a map representing fieldName in POJO:field in Generic Record:Data
+ * type
+ *
+ * @return List of FieldInfo
+ */
+ public List<FieldInfo> createFieldInfoMap(String str)
--- End diff --
Can this method be made private?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by shubham-pathak22 <gi...@git.apache.org>.
Github user shubham-pathak22 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r57716067
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,305 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.ParquetEncodingException;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
+ * POJO class
+ */
+ protected transient Class<?> pojoClass;
+ /**
+ * Map containing setters for fields in POJO
+ */
+ protected transient Map<String, Setter> pojoSetters;
+ /**
+ * String representing Parquet TO POJO field mapping. If not provided, then
+ * reflection is used to determine the mapping. Format :
+ * PARQUET_FIELD_NAME:POJO_FIELD_NAME:TYPE
+ * E.g.event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:
+ * long_id_v2:
+ * LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:
+ * FLOAT,double_val:double_val_v2:DOUBLE
+ */
+ protected transient String groupToPOJOFieldsMapping = null;
+ protected transient List<FieldInfo> fieldInfos;
+ protected transient List<ActiveFieldInfo> columnFieldSetters = null;
+ protected static final String FIELD_SEPARATOR = ":";
+ protected static final String RECORD_SEPARATOR = ",";
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
+ {
+
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ pojoClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ pojoSetters = Maps.newHashMap();
+ for (Field f : pojoClass.getDeclaredFields()) {
+ try {
+ pojoSetters.put(f.getName(), generateSettersForField(pojoClass, f.getName()));
+ } catch (NoSuchFieldException | SecurityException e) {
+ e.printStackTrace();
+ }
+ }
+
+ if (groupToPOJOFieldsMapping == null) {
+ fieldInfos = createFieldInfoMap(generateFieldInfoInputs());
+ } else {
+ fieldInfos = createFieldInfoMap(groupToPOJOFieldsMapping);
+ }
+ initColumnFieldSetters();
+ }
+
+ };
+
+ /**
+ * Converts Group to POJO
+ */
+ @Override
+ protected Object convertGroup(Group group)
+ {
+ Object obj;
+ try {
+ obj = pojoClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException ex) {
+ throw new RuntimeException(ex);
+ }
+ for (int i = 0; i < columnFieldSetters.size(); i++) {
+ ParquetFilePOJOReader.ActiveFieldInfo afi = columnFieldSetters.get(i);
+
--- End diff --
At present, Schema evolution in parquet supports adding new columns. In our case, if new columns are added then user would need to update both, the parquet schema and pojo class. Adding a try/catch does not help in that case.
However, I there might be a case where there are more fields in POJO as compared to schema and that would throw InvalidRecordException. I have added try/catch to handle it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by devtagare <gi...@git.apache.org>.
Github user devtagare commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r57661315
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/AbstractParquetFileReader.java ---
@@ -0,0 +1,113 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.AbstractFileInputOperator;
+
+import parquet.example.data.Group;
+import parquet.hadoop.ParquetReader;
+import parquet.hadoop.example.GroupReadSupport;
+import parquet.schema.MessageType;
+import parquet.schema.MessageTypeParser;
+
+/**
+ * Base implementation of ParquetFileReader. Reads parquet files from input
+ * directory using GroupReadSupport. Derived classes need to implement
+ * {@link #convertGroup(Group)} method to convert Group to other type. Example
+ * of such implementation is {@link ParquetFilePOJOReader} that converts Group
+ * to POJO.
+ *
+ * @since 3.3.3
+ */
+public abstract class AbstractParquetFileReader<T> extends AbstractFileInputOperator<T>
+{
+ private transient ParquetReader<Group> reader;
--- End diff --
Please add a Logger
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58046455
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.InvalidRecordException;
+import parquet.io.ParquetEncodingException;
+import parquet.schema.PrimitiveType.PrimitiveTypeName;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * Parquet primitive types maps to POJO datatypes as follows : INT32 -> int
+ * INT64 -> long BOOLEAN -> boolean FLOAT -> float DOUBLE -> double BINARY ->
+ * String
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
+ * POJO class
+ */
+ protected transient Class<?> pojoClass;
+ /**
+ * Map containing setters for fields in POJO
+ */
+ protected transient Map<String, Setter> pojoSetters;
+ /**
+ * String representing Parquet TO POJO field mapping. If not provided, then
+ * reflection is used to determine the mapping. Format :
+ * PARQUET_FIELD_NAME:POJO_FIELD_NAME:TYPE
+ * E.g.event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:
+ * long_id_v2:
+ * LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:
+ * FLOAT,double_val:double_val_v2:DOUBLE
+ */
+ protected transient String groupToPOJOFieldsMapping = null;
+ protected transient List<FieldInfo> fieldInfos;
+ protected transient List<ActiveFieldInfo> columnFieldSetters = null;
+ protected static final String FIELD_SEPARATOR = ":";
+ protected static final String RECORD_SEPARATOR = ",";
+ private static final Logger logger = LoggerFactory.getLogger(ParquetFilePOJOReader.class);
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
+ {
+
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ pojoClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ pojoSetters = Maps.newHashMap();
+ for (Field f : pojoClass.getDeclaredFields()) {
+ try {
+ pojoSetters.put(f.getName(), generateSettersForField(pojoClass, f.getName()));
+ } catch (NoSuchFieldException | SecurityException e) {
+ logger.error("Failed to generate setters. Exception {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ if (groupToPOJOFieldsMapping == null) {
+ fieldInfos = createFieldInfoMap(generateFieldInfoInputs());
+ } else {
+ fieldInfos = createFieldInfoMap(groupToPOJOFieldsMapping);
+ }
+ initColumnFieldSetters();
+ }
+
+ };
+
+ /**
+ * Converts Group to POJO
+ */
+ @Override
+ protected Object convertGroup(Group group)
+ {
+ Object obj;
+ try {
+ obj = pojoClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException ex) {
+ throw new RuntimeException(ex);
+ }
+ for (int i = 0; i < columnFieldSetters.size(); i++) {
+ try {
+ ParquetFilePOJOReader.ActiveFieldInfo afi = columnFieldSetters.get(i);
+ int fieldIndex = schema.getFieldIndex(afi.fieldInfo.getColumnName());
--- End diff --
With above aproach, you even don't need to get fieldIndex using schema.getFieldIndex everytime.
Your array structure has it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58044050
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.InvalidRecordException;
+import parquet.io.ParquetEncodingException;
+import parquet.schema.PrimitiveType.PrimitiveTypeName;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * Parquet primitive types maps to POJO datatypes as follows : INT32 -> int
+ * INT64 -> long BOOLEAN -> boolean FLOAT -> float DOUBLE -> double BINARY ->
+ * String
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
+ * POJO class
+ */
+ protected transient Class<?> pojoClass;
+ /**
+ * Map containing setters for fields in POJO
+ */
+ protected transient Map<String, Setter> pojoSetters;
+ /**
+ * String representing Parquet TO POJO field mapping. If not provided, then
+ * reflection is used to determine the mapping. Format :
+ * PARQUET_FIELD_NAME:POJO_FIELD_NAME:TYPE
+ * E.g.event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:
+ * long_id_v2:
+ * LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:
+ * FLOAT,double_val:double_val_v2:DOUBLE
+ */
+ protected transient String groupToPOJOFieldsMapping = null;
+ protected transient List<FieldInfo> fieldInfos;
+ protected transient List<ActiveFieldInfo> columnFieldSetters = null;
+ protected static final String FIELD_SEPARATOR = ":";
+ protected static final String RECORD_SEPARATOR = ",";
+ private static final Logger logger = LoggerFactory.getLogger(ParquetFilePOJOReader.class);
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
+ {
+
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ pojoClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ pojoSetters = Maps.newHashMap();
+ for (Field f : pojoClass.getDeclaredFields()) {
+ try {
+ pojoSetters.put(f.getName(), generateSettersForField(pojoClass, f.getName()));
+ } catch (NoSuchFieldException | SecurityException e) {
+ logger.error("Failed to generate setters. Exception {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ if (groupToPOJOFieldsMapping == null) {
+ fieldInfos = createFieldInfoMap(generateFieldInfoInputs());
+ } else {
+ fieldInfos = createFieldInfoMap(groupToPOJOFieldsMapping);
+ }
+ initColumnFieldSetters();
+ }
+
+ };
+
+ /**
+ * Converts Group to POJO
+ */
+ @Override
+ protected Object convertGroup(Group group)
+ {
+ Object obj;
+ try {
+ obj = pojoClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException ex) {
+ throw new RuntimeException(ex);
+ }
+ for (int i = 0; i < columnFieldSetters.size(); i++) {
+ try {
+ ParquetFilePOJOReader.ActiveFieldInfo afi = columnFieldSetters.get(i);
+ int fieldIndex = schema.getFieldIndex(afi.fieldInfo.getColumnName());
+ PrimitiveTypeName primitiveTypeName = schema.getType(fieldIndex).asPrimitiveType().getPrimitiveTypeName();
+
+ switch (primitiveTypeName) {
+
+ case BOOLEAN:
+ Boolean booleanVal = Boolean.parseBoolean(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, booleanVal);
+ break;
+
+ case INT32:
+ Integer intVal = Integer.parseInt(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, intVal);
+ break;
+
+ case INT64:
+ Long longVal = Long.parseLong(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, longVal);
+ break;
+
+ case FLOAT:
+ Float floatVal = Float.parseFloat(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, floatVal);
+ break;
+
+ case DOUBLE:
+ Double doubleVal = Double.parseDouble(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, doubleVal);
+ break;
+
+ case BINARY:
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, group.getValueToString(fieldIndex, 0));
+ break;
+
+ default:
+ throw new ParquetEncodingException("Unsupported column type: " + primitiveTypeName);
+
+ }
+ } catch (InvalidRecordException e) {
+ logger.error("Field not found in schema {} ", e);
+ }
+ }
+ return obj;
+ }
+
+ @Override
+ protected void emit(Object tuple)
+ {
+ output.emit(tuple);
+ }
+
+ /**
+ * Sets the {@link FieldInfo}s. A {@link FieldInfo} maps a store column to a
+ * pojo field name.<br/>
+ * The value from fieldInfo.column is assigned to
+ * fieldInfo.pojoFieldExpression.
+ *
+ * @description $[].columnName name of the Output Field in POJO
+ * @description $[].pojoFieldExpression expression to get the respective field
+ * from generic record
+ * @useSchema $[].pojoFieldExpression outputPort.fields[].name
+ */
+ public void setFieldInfos(List<FieldInfo> fieldInfos)
+ {
+ this.fieldInfos = fieldInfos;
+ }
+
+ /**
+ * Creates a map representing fieldName in POJO:field in Generic Record:Data
+ * type
+ *
+ * @return List of FieldInfo
+ */
+ public List<FieldInfo> createFieldInfoMap(String str)
+ {
+ fieldInfos = new ArrayList<FieldInfo>();
+
+ StringTokenizer strtok = new StringTokenizer(str, RECORD_SEPARATOR);
+
+ while (strtok.hasMoreTokens()) {
+ String[] token = strtok.nextToken().split(FIELD_SEPARATOR);
+
+ fieldInfos.add(new FieldInfo(token[0], token[1], SupportType.valueOf(token[2])));
+ }
+
+ return fieldInfos;
+ }
+
+ /**
+ * Generates setter for given field and POJO class
+ *
+ * @param klass
+ * @param inputFieldName
+ * @return Setter
+ * @throws NoSuchFieldException
+ * @throws SecurityException
+ */
+ private Setter generateSettersForField(Class<?> klass, String inputFieldName)
+ throws NoSuchFieldException, SecurityException
+ {
+ Field f = klass.getDeclaredField(inputFieldName);
+ Class c = ClassUtils.primitiveToWrapper(f.getType());
+ Setter classSetter = PojoUtils.createSetter(klass, inputFieldName, c);
+ return classSetter;
+ }
+
+ /**
+ * Returns groupToPOJOFieldsMapping
+ *
+ * @return groupToPOJOFieldsMapping
+ */
+ public String getGroupToPOJOFieldsMapping()
+ {
+ return groupToPOJOFieldsMapping;
+ }
+
+ /**
+ * Sets groupToPOJOFieldsMapping
+ *
+ * @param groupToPOJOFieldsMapping
+ */
+ public void setGroupToPOJOFieldsMapping(String groupToPOJOFieldsMapping)
+ {
+ this.groupToPOJOFieldsMapping = groupToPOJOFieldsMapping;
+ }
+
+ /**
+ * Class that maps fieldInfo to its getters or setters
+ */
+ protected static class ActiveFieldInfo
+ {
+ final FieldInfo fieldInfo;
+ Object setterOrGetter;
--- End diff --
setterOrGetter seems to be unused... Can this be removed.. And as fieldInfo is a single object in this class... Do we really need this class?
Also there is a List<FieldInfo> variable... How is it seperate from this one? If its meant to serve same purpose, can they be combined?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58045703
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.InvalidRecordException;
+import parquet.io.ParquetEncodingException;
+import parquet.schema.PrimitiveType.PrimitiveTypeName;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * Parquet primitive types maps to POJO datatypes as follows : INT32 -> int
+ * INT64 -> long BOOLEAN -> boolean FLOAT -> float DOUBLE -> double BINARY ->
+ * String
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
+ * POJO class
+ */
+ protected transient Class<?> pojoClass;
+ /**
+ * Map containing setters for fields in POJO
+ */
+ protected transient Map<String, Setter> pojoSetters;
+ /**
+ * String representing Parquet TO POJO field mapping. If not provided, then
+ * reflection is used to determine the mapping. Format :
+ * PARQUET_FIELD_NAME:POJO_FIELD_NAME:TYPE
+ * E.g.event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:
+ * long_id_v2:
+ * LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:
+ * FLOAT,double_val:double_val_v2:DOUBLE
+ */
+ protected transient String groupToPOJOFieldsMapping = null;
+ protected transient List<FieldInfo> fieldInfos;
+ protected transient List<ActiveFieldInfo> columnFieldSetters = null;
--- End diff --
I don't see columnFieldSetters been any dfferent from fieldInfos variable. Can they be merged to single variable?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58042862
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/AbstractParquetFileReader.java ---
@@ -0,0 +1,113 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.AbstractFileInputOperator;
+
+import parquet.example.data.Group;
+import parquet.hadoop.ParquetReader;
+import parquet.hadoop.example.GroupReadSupport;
+import parquet.schema.MessageType;
+import parquet.schema.MessageTypeParser;
+
+/**
+ * Base implementation of ParquetFileReader. Reads parquet files from input
+ * directory using GroupReadSupport. Derived classes need to implement
+ * {@link #convertGroup(Group)} method to convert Group to other type. Example
+ * of such implementation is {@link ParquetFilePOJOReader} that converts Group
+ * to POJO.
+ *
+ * @since 3.3.3
--- End diff --
Please remove @since tag
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by shubham-pathak22 <gi...@git.apache.org>.
Github user shubham-pathak22 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r57717541
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,305 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.ParquetEncodingException;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
--- End diff --
Done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by shubham-pathak22 <gi...@git.apache.org>.
Github user shubham-pathak22 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58029916
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,305 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.ParquetEncodingException;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
+ * POJO class
+ */
+ protected transient Class<?> pojoClass;
+ /**
+ * Map containing setters for fields in POJO
+ */
+ protected transient Map<String, Setter> pojoSetters;
+ /**
+ * String representing Parquet TO POJO field mapping. If not provided, then
+ * reflection is used to determine the mapping. Format :
+ * PARQUET_FIELD_NAME:POJO_FIELD_NAME:TYPE
+ * E.g.event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:
+ * long_id_v2:
+ * LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:
+ * FLOAT,double_val:double_val_v2:DOUBLE
+ */
+ protected transient String groupToPOJOFieldsMapping = null;
+ protected transient List<FieldInfo> fieldInfos;
+ protected transient List<ActiveFieldInfo> columnFieldSetters = null;
+ protected static final String FIELD_SEPARATOR = ":";
+ protected static final String RECORD_SEPARATOR = ",";
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
+ {
+
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ pojoClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ pojoSetters = Maps.newHashMap();
+ for (Field f : pojoClass.getDeclaredFields()) {
+ try {
+ pojoSetters.put(f.getName(), generateSettersForField(pojoClass, f.getName()));
+ } catch (NoSuchFieldException | SecurityException e) {
+ e.printStackTrace();
+ }
+ }
+
+ if (groupToPOJOFieldsMapping == null) {
+ fieldInfos = createFieldInfoMap(generateFieldInfoInputs());
+ } else {
+ fieldInfos = createFieldInfoMap(groupToPOJOFieldsMapping);
+ }
+ initColumnFieldSetters();
+ }
+
+ };
+
+ /**
+ * Converts Group to POJO
+ */
+ @Override
+ protected Object convertGroup(Group group)
+ {
+ Object obj;
+ try {
+ obj = pojoClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException ex) {
+ throw new RuntimeException(ex);
+ }
+ for (int i = 0; i < columnFieldSetters.size(); i++) {
+ ParquetFilePOJOReader.ActiveFieldInfo afi = columnFieldSetters.get(i);
+
+ afi = columnFieldSetters.get(i);
+ int fieldIndex = schema.getFieldIndex(afi.fieldInfo.getColumnName());
+ SupportType st = afi.fieldInfo.getType();
+
+ switch (st) {
+
+ case BOOLEAN:
+ Boolean booleanVal = Boolean.parseBoolean(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, booleanVal);
+ break;
+
+ case INTEGER:
+ Integer intVal = Integer.parseInt(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, intVal);
+ break;
+
+ case LONG:
+ Long longVal = Long.parseLong(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, longVal);
+ break;
+
+ case FLOAT:
+ Float floatVal = Float.parseFloat(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, floatVal);
+ break;
+
+ case DOUBLE:
+ Double doubleVal = Double.parseDouble(group.getValueToString(fieldIndex, 0));
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, doubleVal);
+ break;
+
+ case STRING:
+ pojoSetters.get(afi.fieldInfo.getPojoFieldExpression()).set(obj, group.getValueToString(fieldIndex, 0));
+ break;
+
+ default:
+ throw new ParquetEncodingException("Unsupported column type: " + schema.getType(fieldIndex));
--- End diff --
This would require adding support for handling byte arrays in POJOUtils. We can enhance this operator later.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r58043182
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/AbstractParquetFileReader.java ---
@@ -0,0 +1,113 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.AbstractFileInputOperator;
+
+import parquet.example.data.Group;
+import parquet.hadoop.ParquetReader;
+import parquet.hadoop.example.GroupReadSupport;
+import parquet.schema.MessageType;
+import parquet.schema.MessageTypeParser;
+
+/**
+ * Base implementation of ParquetFileReader. Reads parquet files from input
+ * directory using GroupReadSupport. Derived classes need to implement
+ * {@link #convertGroup(Group)} method to convert Group to other type. Example
+ * of such implementation is {@link ParquetFilePOJOReader} that converts Group
+ * to POJO.
+ *
+ * @since 3.3.3
+ */
+public abstract class AbstractParquetFileReader<T> extends AbstractFileInputOperator<T>
+{
+ private transient ParquetReader<Group> reader;
+ protected transient MessageType schema;
+ protected String parquetSchema;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ schema = MessageTypeParser.parseMessageType(parquetSchema);
+ }
+
+ /**
+ * Opens the file to read using GroupReadSupport
+ */
+ @Override
+ protected InputStream openFile(Path path) throws IOException
+ {
+ InputStream is = super.openFile(path);
+ GroupReadSupport readSupport = new GroupReadSupport();
+ readSupport.init(configuration, null, schema);
+ reader = new ParquetReader<Group>(path, readSupport);
+ return is;
+ }
+
+ /**
+ * Reads next record in parquet file as a group. Returns null when end of file
+ * is reached
+ */
+ @Override
+ protected T readEntity() throws IOException
+ {
+ Group group = reader.read();
+ if (group != null) {
+ return convertGroup(group);
+ }
+ return null;
+ }
+
+ /**
+ * Converts Group to object of type T. Derived classes need to provide
+ * implementation for conversion
+ *
+ * @param Group
--- End diff --
g should be lower cased.. "group" instead of "Group".
Also can you add description for this param.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2014 added parquet ...
Posted by devtagare <gi...@git.apache.org>.
Github user devtagare commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/219#discussion_r57660735
--- Diff: contrib/src/main/java/com/datatorrent/contrib/parquet/ParquetFilePOJOReader.java ---
@@ -0,0 +1,305 @@
+/**
+ * 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 com.datatorrent.contrib.parquet;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.apache.commons.lang3.ClassUtils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+
+import parquet.example.data.Group;
+import parquet.io.ParquetEncodingException;
+
+/**
+ * <p>
+ * ParquetFilePOJOReader
+ * </p>
+ * ParquetFilePOJOReader operator is a concrete implementation of
+ * AbstractParquetFileReader to read Parquet files and emit records as POJOs.The
+ * POJO class name & field mapping should be provided by the user. If this
+ * mapping is not provided then reflection is used to determine this mapping. As
+ * of now only primitive types ( INT32, INT64, BOOLEAN, FLOAT, DOUBLE, BINARY )
+ * are supported.
+ *
+ * @displayName ParquetFilePOJOReader
+ * @tags parquet,input adapter
+ * @since 3.3.0
+ */
+public class ParquetFilePOJOReader extends AbstractParquetFileReader<Object>
+{
+
+ /**
+ * POJO class
+ */
+ protected transient Class<?> pojoClass;
+ /**
+ * Map containing setters for fields in POJO
+ */
+ protected transient Map<String, Setter> pojoSetters;
+ /**
+ * String representing Parquet TO POJO field mapping. If not provided, then
+ * reflection is used to determine the mapping. Format :
+ * PARQUET_FIELD_NAME:POJO_FIELD_NAME:TYPE
+ * E.g.event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:
+ * long_id_v2:
+ * LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:
+ * FLOAT,double_val:double_val_v2:DOUBLE
+ */
+ protected transient String groupToPOJOFieldsMapping = null;
+ protected transient List<FieldInfo> fieldInfos;
+ protected transient List<ActiveFieldInfo> columnFieldSetters = null;
+ protected static final String FIELD_SEPARATOR = ":";
+ protected static final String RECORD_SEPARATOR = ",";
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
+ {
+
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ pojoClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ pojoSetters = Maps.newHashMap();
+ for (Field f : pojoClass.getDeclaredFields()) {
+ try {
+ pojoSetters.put(f.getName(), generateSettersForField(pojoClass, f.getName()));
+ } catch (NoSuchFieldException | SecurityException e) {
+ e.printStackTrace();
+ }
+ }
+
+ if (groupToPOJOFieldsMapping == null) {
+ fieldInfos = createFieldInfoMap(generateFieldInfoInputs());
+ } else {
+ fieldInfos = createFieldInfoMap(groupToPOJOFieldsMapping);
+ }
+ initColumnFieldSetters();
+ }
+
+ };
+
+ /**
+ * Converts Group to POJO
+ */
+ @Override
+ protected Object convertGroup(Group group)
+ {
+ Object obj;
+ try {
+ obj = pojoClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException ex) {
+ throw new RuntimeException(ex);
+ }
+ for (int i = 0; i < columnFieldSetters.size(); i++) {
+ ParquetFilePOJOReader.ActiveFieldInfo afi = columnFieldSetters.get(i);
+
--- End diff --
Adding a try/catch here would help.Exception in setting one field should not cause the group to fail since parquet supports schema evolution.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---