Abacn commented on code in PR #32695:
URL: https://github.com/apache/beam/pull/32695#discussion_r1793766780
##########
sdks/java/io/hcatalog/src/main/java/org/apache/beam/sdk/io/hcatalog/HCatToRow.java:
##########
@@ -74,14 +77,23 @@ public PCollection<Row> expand(PBegin input) {
private static class HCatToRowFn extends DoFn<HCatRecord, Row> {
private final Schema schema;
+ private Object castHDate(Object obj) {
+ if (obj instanceof org.apache.hadoop.hive.common.type.Date) {
+ return new Instant(((org.apache.hadoop.hive.common.type.Date)
obj).toEpochMilli());
+ }
+ return obj;
+ }
+
HCatToRowFn(Schema schema) {
this.schema = schema;
}
@ProcessElement
public void processElement(ProcessContext c) {
HCatRecord hCatRecord = c.element();
- c.output(Row.withSchema(schema).addValues(hCatRecord.getAll()).build());
+ List<Object> recordValues =
+
hCatRecord.getAll().stream().map(this::castHDate).collect(Collectors.toList());
Review Comment:
Shall we generalize the function, e.g. just naming as "castTypes", leaving
space for future improvement if other type need a conversion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
