You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2020/09/29 19:51:08 UTC
[GitHub] [hudi] tandonraghavs opened a new issue #2131: HUDI with Mongo Oplogs (Debezium)
tandonraghavs opened a new issue #2131:
URL: https://github.com/apache/hudi/issues/2131
We are planning to use HUDI for our Data Warehouse, and we want to dump Mongo Data to S3.
For Mongo we are relying on Oplogs(Debezium) . But as I experimented with Hudi, Hudi is not working if I have partial Data(oplogs kind of data).
Is it recommended to have entire Row for Hudi?
- I also tried to overwrite **HoodieRecordPayload** for **combineAndGetUpdateValue**, but that is also not fulfilling the use case as Compaction is running in Background then it only compares the recent record and stored record. So, few of the oplogs updates are getting lost.
- There is another issue in Oplogs where we can have multiple Ids of same Document, then in this case I want to apply update using all the records, but due to preCombine it is taking updates of only latest record.
- Also, there is no option to set **hoodie.combine.before.upsert=false** while using DataSource.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] tandonraghavs commented on issue #2131: [SUPPORT] HUDI with Mongo Oplogs (Debezium)
Posted by GitBox <gi...@apache.org>.
tandonraghavs commented on issue #2131:
URL: https://github.com/apache/hudi/issues/2131#issuecomment-703137335
@bvaradar Thanks for the PR.
I tried the way you suggested and it is working as desired.
Let me know if you see any issue in below code.
````
public class MergeHudiPayload implements HoodieRecordPayload<MergeHudiPayload> {
private byte[] recordBytes;
private Schema schema;
private Comparable orderingVal;
public MergeHudiPayload(GenericRecord genericRecord, Comparable orderingVal) {
try {
if (genericRecord!=null) {
this.recordBytes = HoodieAvroUtils.avroToBytes(genericRecord);
this.schema=genericRecord.getSchema();
this.orderingVal=orderingVal;
} else {
this.recordBytes = new byte[0];
}
} catch (IOException io) {
throw new HoodieIOException("Cannot convert record to bytes", io);
}
}
//Constructor called during Compaction
public MergeHudiPayload(Option<GenericRecord> record) {
this(record.isPresent() ? record.get() : null, (record1) -> 0); // natural order
}
}
````
Then I use it like this in **preCombine**
````
GenericRecord currentRecord=HoodieAvroUtils.bytesToAvro(currentObj.recordBytes,currentObj.schema);
GenericRecord anotherRecord=HoodieAvroUtils.bytesToAvro(anotherObj.recordBytes,anotherObj.schema);
````
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] tandonraghavs edited a comment on issue #2131: [SUPPORT] HUDI with Mongo Oplogs (Debezium)
Posted by GitBox <gi...@apache.org>.
tandonraghavs edited a comment on issue #2131:
URL: https://github.com/apache/hudi/issues/2131#issuecomment-702082811
@bvaradar I am using my Custom class as `PAYLOAD_CLASS_OPT_KEY` key -> But the problem is **preCombine** doesnt have reference to **Schema** and it is giving me bytes , so how do i get the Generic Record out of it?
Which is the reason I am not able to implement any custom logic in _preCombine_ as I did in _combineAndGetUpdateValue_.
I am using hudi via Spark Datasource (0.5.3).
And due to the scale of data I dont want to run Compaction after every commit, so using _INLINE_COMPACT_NUM_DELTA_COMMITS_PROP_
- How do i get hold of Schema in preCombine?
Sample Code of my Spark job.
**jsonDf** -> This is a simple Json String which contains the resords.
````
Dataset<GenericRecord> data=jsonDf.map((MapFunction<String, GenericRecord>) record ->
generateHoodieRecord(record, schemaStr),Encoders.bean(GenericRecord.class));
Dataset<Row> ds= AvroConversionUtils.createDataFrame(data.rdd(),
schemaStr,sparkSession);
ds
.write().format("org.apache.hudi").
.options ()...
.mode(SaveMode.Append)
.save(tablePath);
````
This Ticket also talks about the same - https://issues.apache.org/jira/browse/HUDI-898
Also, i think we cannot add/remove any field values in preCombine, as doing it manually is causing _EOFException in reading log file._
Am i missing something here? So, i dont think we can use Hudi if we have partial Records coming in for oplogs and we have to apply all the oplogs to the existing dataset.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] tandonraghavs commented on issue #2131: [SUPPORT] HUDI with Mongo Oplogs (Debezium)
Posted by GitBox <gi...@apache.org>.
tandonraghavs commented on issue #2131:
URL: https://github.com/apache/hudi/issues/2131#issuecomment-702082811
@bvaradar I am using my Custom class as `PAYLOAD_CLASS_OPT_KEY` key -> But the problem is **preCombine** doesnt have reference to **Schema** and it is giving me bytes , so how do i get the Generic Record out of it?
Which is the reason I am not able to implement any custom logic in _preCombine_ as I did in _combineAndGetUpdateValue_.
I am using hudi via Spark Datasource (0.5.3).
And due to the scale of data I dont want to run Compaction after every commit, so using _INLINE_COMPACT_NUM_DELTA_COMMITS_PROP_
- How do i get hold of Schema in preCombine?
Sample Code of my Spark job.
**jsonDf** -> This is a simple Json String which contains the resords.
````
Dataset<GenericRecord> data=jsonDf.map((MapFunction<String, GenericRecord>) record ->
generateHoodieRecord(record, schemaStr),Encoders.bean(GenericRecord.class));
Dataset<Row> ds= AvroConversionUtils.createDataFrame(data.rdd(),
schemaStr,sparkSession);
ds
.write().format("org.apache.hudi").
.options ()...
.mode(SaveMode.Append)
.save(tablePath);
````
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] bvaradar commented on issue #2131: [SUPPORT] HUDI with Mongo Oplogs (Debezium)
Posted by GitBox <gi...@apache.org>.
bvaradar commented on issue #2131:
URL: https://github.com/apache/hudi/issues/2131#issuecomment-703170642
> @bvaradar Thanks for the PR.
> I tried the way you suggested and it is working as desired.
> Let me know if you see any issue in below code.
>
> ```
> public class MergeHudiPayload implements HoodieRecordPayload<MergeHudiPayload> {
> private byte[] recordBytes;
> private Schema schema;
> private Comparable orderingVal;
>
> public MergeHudiPayload(GenericRecord genericRecord, Comparable orderingVal) {
> try {
> if (genericRecord!=null) {
> this.recordBytes = HoodieAvroUtils.avroToBytes(genericRecord);
> this.schema=genericRecord.getSchema();
> this.orderingVal=orderingVal;
> } else {
> this.recordBytes = new byte[0];
> }
> } catch (IOException io) {
> throw new HoodieIOException("Cannot convert record to bytes", io);
> }
> }
>
> //Constructor called during Compaction
> public MergeHudiPayload(Option<GenericRecord> record) {
> this(record.isPresent() ? record.get() : null, (record1) -> 0); // natural order
> }
> }
> ```
>
> Then I use it like this in **preCombine**
>
> ```
> GenericRecord currentRecord=HoodieAvroUtils.bytesToAvro(currentObj.recordBytes,currentObj.schema);
> GenericRecord anotherRecord=HoodieAvroUtils.bytesToAvro(anotherObj.recordBytes,anotherObj.schema);
> ```
Great. This looks good. Although, Please check if you might want to store the schema as a string in the constructors and lazily load. Last I checked Avro Schema Object does not implement serializable and you may run into Serialization errors
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] tandonraghavs edited a comment on issue #2131: [SUPPORT] HUDI with Mongo Oplogs (Debezium)
Posted by GitBox <gi...@apache.org>.
tandonraghavs edited a comment on issue #2131:
URL: https://github.com/apache/hudi/issues/2131#issuecomment-702082811
@bvaradar I am using my Custom class as `PAYLOAD_CLASS_OPT_KEY` key -> But the problem is **preCombine** doesnt have reference to **Schema** and it is giving me bytes , so how do i get the Generic Record out of it?
Which is the reason I am not able to implement any custom logic in _preCombine_ as I did in _combineAndGetUpdateValue_.
I am using hudi via Spark Datasource (0.5.3).
And due to the scale of data I dont want to run Compaction after every commit, so using _INLINE_COMPACT_NUM_DELTA_COMMITS_PROP_
- How do i get hold of Schema in preCombine?
Sample Code of my Spark job.
**jsonDf** -> This is a simple Json String which contains the resords.
````
Dataset<GenericRecord> data=jsonDf.map((MapFunction<String, GenericRecord>) record ->
generateHoodieRecord(record, schemaStr),Encoders.bean(GenericRecord.class));
Dataset<Row> ds= AvroConversionUtils.createDataFrame(data.rdd(),
schemaStr,sparkSession);
ds
.write().format("org.apache.hudi").
.options ()...
.mode(SaveMode.Append)
.save(tablePath);
````
This Ticket also talks about the same - https://issues.apache.org/jira/browse/HUDI-898
- Also, i think we cannot add/remove any field values in preCombine, as doing it manually is causing _EOFException in reading log file._
- Am i missing something here? So, i dont think we can use Hudi if we have partial Records coming in for oplogs and we have to apply all the oplogs to the existing dataset.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] bvaradar commented on issue #2131: [SUPPORT] HUDI with Mongo Oplogs (Debezium)
Posted by GitBox <gi...@apache.org>.
bvaradar commented on issue #2131:
URL: https://github.com/apache/hudi/issues/2131#issuecomment-702846845
@tandonraghavs : Thanks for the explanation. I have created a PR to expose Schema in preCombine : https://github.com/apache/hudi/pull/2141
There is one option for you which is to implement your RecordPayload such that it does not inherit from BaseAvroPayload or inherit it but in your constructor, hold the schema of the generic record getting passed. You can then use this to implement precombine. Can you give this a try ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] tandonraghavs commented on issue #2131: [SUPPORT] HUDI with Mongo Oplogs (Debezium)
Posted by GitBox <gi...@apache.org>.
tandonraghavs commented on issue #2131:
URL: https://github.com/apache/hudi/issues/2131#issuecomment-703275211
@bvaradar Thanks for the input. You are right since i tested in local it was working fine...But i converted Schema-> byte[].
Closing this ticket as things are working fine.
Thanks for the support!!!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] tandonraghavs closed issue #2131: [SUPPORT] HUDI with Mongo Oplogs (Debezium)
Posted by GitBox <gi...@apache.org>.
tandonraghavs closed issue #2131:
URL: https://github.com/apache/hudi/issues/2131
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] bvaradar commented on issue #2131: [SUPPORT] HUDI with Mongo Oplogs (Debezium)
Posted by GitBox <gi...@apache.org>.
bvaradar commented on issue #2131:
URL: https://github.com/apache/hudi/issues/2131#issuecomment-702072024
@tandonraghavs : For compaction, the payload class defined in hoodie.properties is used for pre-combining. Can you check what is the payload class configured in hoodie.properties ?
As you would have noticed, compaction would first pre-combines all delta records first before merging using combineAndGetUpdateValue. So, you would need to implement custom merging in preCombine method in your payload class.
IIUC, Oplogs are partial row image(updates). Wouldn't it be possible to employ the same merging logic that you were doing as part of combineAndGetUpdateValue ?
Also, which version of Hudi are you using ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] tandonraghavs edited a comment on issue #2131: [SUPPORT] HUDI with Mongo Oplogs (Debezium)
Posted by GitBox <gi...@apache.org>.
tandonraghavs edited a comment on issue #2131:
URL: https://github.com/apache/hudi/issues/2131#issuecomment-702082811
@bvaradar I am using my Custom class as `PAYLOAD_CLASS_OPT_KEY` key -> But the problem is **preCombine** doesnt have reference to **Schema** and it is giving me bytes , so how do i get the Generic Record out of it?
Which is the reason I am not able to implement any custom logic in _preCombine_ as I did in _combineAndGetUpdateValue_.
I am using hudi via Spark Datasource (0.5.3).
And due to the scale of data I dont want to run Compaction after every commit, so using _INLINE_COMPACT_NUM_DELTA_COMMITS_PROP_
- How do i get hold of Schema in preCombine?
Sample Code of my Spark job.
**jsonDf** -> This is a simple Json String which contains the resords.
````
Dataset<GenericRecord> data=jsonDf.map((MapFunction<String, GenericRecord>) record ->
generateHoodieRecord(record, schemaStr),Encoders.bean(GenericRecord.class));
Dataset<Row> ds= AvroConversionUtils.createDataFrame(data.rdd(),
schemaStr,sparkSession);
ds
.write().format("org.apache.hudi").
.options ()...
.mode(SaveMode.Append)
.save(tablePath);
````
This Ticket also talks about the same - https://issues.apache.org/jira/browse/HUDI-898
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org