You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@nifi.apache.org by edi mari <ed...@gmail.com> on 2024/03/12 08:38:00 UTC
MergeRecord takes too much time
Hi,
My task is to query Elastic and save the results in a Parquet file.
I'm querying Elastic using the *PaginatedJsonQueryElasticsearch* processor.
The files coming from Elastic are in JSON format, and I'm using the
*MergeRecord* processor to convert the JSON to Parquet format and merge the
result into one file .
The *Record Reader* uses the *JsonTreeReader* and *Record Writer* uses the
*ParquetRecordSetWriter* controller, which uses AVRO schema (Schema Text)
to help with the conversion task.
The process works fine, the only problem is that it takes too much time.
Converting 5 MB takes 30 minutes.
Do you have any idea how to enhance the process?
[image: image.png]
[image: image.png]
[image: image.png]
Re: MergeRecord takes too much time
Posted by Matt Burgess <ma...@apache.org>.
IMO, Best Practice when using Infer Schema is to have the first Reader in
the flow infer the schema, then configure your writers to write the schema
in an attribute and all the rest of the readers to use that attribute. It
avoids performing schema inference more than once. The exception to this
pattern is if you have a large number of fields which could cause the
schema to be too large to fit in an attribute. In that case I recommend
using Avro with embedded schema as the intermediate format throughout the
flow (until it is converted for the target system).
Regards,
Matt
On Mon, Mar 18, 2024 at 9:44 AM Mike Thomsen <mi...@gmail.com> wrote:
> Based on your screenshots, I think you may also be running into a problem
> with running schema inference on exceptionally large inputs. I never use
> schema inference personally, but IIRC one of the things it does in use
> cases like yours is attempt to figure out if it can reuse an existing
> generated schema. For say... 50kb of JSON that's not big deal. 1GB spread
> over 74 flow files, however, could be very computationally significant
> depending on what those documents look like. I would recommend looking at
> the official Avro documentation and trying out some basic hand-written
> schemas for data that large. One of the nice things about that is that you
> can write a schema that only covers the fields you want to keep (convert
> record should be able to remove all of the undefined, extraneous fields). I
> did that with a large data set where the files were frequently 5mb-10mb of
> gnarly, nasty JSON and NiFi's parquet writer tore through them quickly.
>
> On Mon, Mar 18, 2024 at 5:30 AM edi mari <ed...@gmail.com> wrote:
>
>> Thank you for your response, Chris.
>> I tried your suggestion and used the ConvertRecord processor.
>> The performance was better, but the process is still slow. As you can see
>> from the screenshot below, it's taking 5 minutes to process 166 MB.
>>
>>
>> [image: image.png]
>>
>> [image: image.png]
>>
>> [image: image.png]
>>
>> [image: image.png]
>> [image: image.png]
>>
>> Thanks
>> Edi
>>
>> On Tue, Mar 12, 2024 at 12:20 PM Chris Sampson <ch...@naimuri.com>
>> wrote:
>>
>>> I’m not an expert with MergeRecord, but looking at your screenshots, I’d
>>> guess that your setup is taking that long to reach one of the defined
>>> “maximum” settings, e.g. 2GB, 5,000,000 records, or 3600 seconds (1 hour).
>>>
>>> How large (number of records and content size in bytes) are the typical
>>> FlowFiles you’re sending to Merge Record, and how often do they arrive? For
>>> example, if you’re getting results form Elasticsearch in chunks of 10,000
>>> records per response and that’s 10MB in size every 1 second, it’s going to
>>> take a long time to meet any of the defined maximums.
>>>
>>> How have you configured the Elasticsearch processor and what are you
>>> trying to combine together in your flow? Is it that you’re outputting a
>>> single FlowFile per Response (the default setting for “Search Results
>>> Split”), then trying to merge together all responses from a single query
>>> into one FlowFile? If so, I’d suggest changing the Elasticsearch
>>> processor’s “Search Results Split” to be “Per Query” instead, and increase
>>> the “Size” setting (leaving this blank will use the Elasticsearch default
>>> page size, which is often set as “10”). You might then be able to avoid the
>>> need for MergeRecord at all and the conversion of JSON to Parquet could be
>>> done with a ConvertRecord processor instead, for example.
>>>
>>>
>>> Cheers,
>>>
>>> ---
>>> Chris Sampson
>>> IT Consultant
>>> chris.sampson@naimuri.com
>>>
>>>
>>> On 12 Mar 2024, at 08:38, edi mari <ed...@gmail.com> wrote:
>>>
>>> Hi,
>>> My task is to query Elastic and save the results in a Parquet file.
>>> I'm querying Elastic using the *PaginatedJsonQueryElasticsearch*
>>> processor.
>>> The files coming from Elastic are in JSON format, and I'm using the
>>> *MergeRecord* processor to convert the JSON to Parquet format and merge
>>> the result into one file .
>>> The *Record Reader* uses the *JsonTreeReader* and *Record Writer* uses
>>> the *ParquetRecordSetWriter* controller, which uses AVRO schema (Schema
>>> Text) to help with the conversion task.
>>>
>>> The process works fine, the only problem is that it takes too much time.
>>> Converting 5 MB takes 30 minutes.
>>>
>>> Do you have any idea how to enhance the process?
>>>
>>> <image.png>
>>>
>>> <image.png>
>>> <image.png>
>>>
>>>
>>>
Re: MergeRecord takes too much time
Posted by Mike Thomsen <mi...@gmail.com>.
Based on your screenshots, I think you may also be running into a problem
with running schema inference on exceptionally large inputs. I never use
schema inference personally, but IIRC one of the things it does in use
cases like yours is attempt to figure out if it can reuse an existing
generated schema. For say... 50kb of JSON that's not big deal. 1GB spread
over 74 flow files, however, could be very computationally significant
depending on what those documents look like. I would recommend looking at
the official Avro documentation and trying out some basic hand-written
schemas for data that large. One of the nice things about that is that you
can write a schema that only covers the fields you want to keep (convert
record should be able to remove all of the undefined, extraneous fields). I
did that with a large data set where the files were frequently 5mb-10mb of
gnarly, nasty JSON and NiFi's parquet writer tore through them quickly.
On Mon, Mar 18, 2024 at 5:30 AM edi mari <ed...@gmail.com> wrote:
> Thank you for your response, Chris.
> I tried your suggestion and used the ConvertRecord processor.
> The performance was better, but the process is still slow. As you can see
> from the screenshot below, it's taking 5 minutes to process 166 MB.
>
>
> [image: image.png]
>
> [image: image.png]
>
> [image: image.png]
>
> [image: image.png]
> [image: image.png]
>
> Thanks
> Edi
>
> On Tue, Mar 12, 2024 at 12:20 PM Chris Sampson <ch...@naimuri.com>
> wrote:
>
>> I’m not an expert with MergeRecord, but looking at your screenshots, I’d
>> guess that your setup is taking that long to reach one of the defined
>> “maximum” settings, e.g. 2GB, 5,000,000 records, or 3600 seconds (1 hour).
>>
>> How large (number of records and content size in bytes) are the typical
>> FlowFiles you’re sending to Merge Record, and how often do they arrive? For
>> example, if you’re getting results form Elasticsearch in chunks of 10,000
>> records per response and that’s 10MB in size every 1 second, it’s going to
>> take a long time to meet any of the defined maximums.
>>
>> How have you configured the Elasticsearch processor and what are you
>> trying to combine together in your flow? Is it that you’re outputting a
>> single FlowFile per Response (the default setting for “Search Results
>> Split”), then trying to merge together all responses from a single query
>> into one FlowFile? If so, I’d suggest changing the Elasticsearch
>> processor’s “Search Results Split” to be “Per Query” instead, and increase
>> the “Size” setting (leaving this blank will use the Elasticsearch default
>> page size, which is often set as “10”). You might then be able to avoid the
>> need for MergeRecord at all and the conversion of JSON to Parquet could be
>> done with a ConvertRecord processor instead, for example.
>>
>>
>> Cheers,
>>
>> ---
>> Chris Sampson
>> IT Consultant
>> chris.sampson@naimuri.com
>>
>>
>> On 12 Mar 2024, at 08:38, edi mari <ed...@gmail.com> wrote:
>>
>> Hi,
>> My task is to query Elastic and save the results in a Parquet file.
>> I'm querying Elastic using the *PaginatedJsonQueryElasticsearch*
>> processor.
>> The files coming from Elastic are in JSON format, and I'm using the
>> *MergeRecord* processor to convert the JSON to Parquet format and merge
>> the result into one file .
>> The *Record Reader* uses the *JsonTreeReader* and *Record Writer* uses
>> the *ParquetRecordSetWriter* controller, which uses AVRO schema (Schema
>> Text) to help with the conversion task.
>>
>> The process works fine, the only problem is that it takes too much time.
>> Converting 5 MB takes 30 minutes.
>>
>> Do you have any idea how to enhance the process?
>>
>> <image.png>
>>
>> <image.png>
>> <image.png>
>>
>>
>>
Re: MergeRecord takes too much time
Posted by edi mari <ed...@gmail.com>.
Thank you for your response, Chris.
I tried your suggestion and used the ConvertRecord processor.
The performance was better, but the process is still slow. As you can see
from the screenshot below, it's taking 5 minutes to process 166 MB.
[image: image.png]
[image: image.png]
[image: image.png]
[image: image.png]
[image: image.png]
Thanks
Edi
On Tue, Mar 12, 2024 at 12:20 PM Chris Sampson <ch...@naimuri.com>
wrote:
> I’m not an expert with MergeRecord, but looking at your screenshots, I’d
> guess that your setup is taking that long to reach one of the defined
> “maximum” settings, e.g. 2GB, 5,000,000 records, or 3600 seconds (1 hour).
>
> How large (number of records and content size in bytes) are the typical
> FlowFiles you’re sending to Merge Record, and how often do they arrive? For
> example, if you’re getting results form Elasticsearch in chunks of 10,000
> records per response and that’s 10MB in size every 1 second, it’s going to
> take a long time to meet any of the defined maximums.
>
> How have you configured the Elasticsearch processor and what are you
> trying to combine together in your flow? Is it that you’re outputting a
> single FlowFile per Response (the default setting for “Search Results
> Split”), then trying to merge together all responses from a single query
> into one FlowFile? If so, I’d suggest changing the Elasticsearch
> processor’s “Search Results Split” to be “Per Query” instead, and increase
> the “Size” setting (leaving this blank will use the Elasticsearch default
> page size, which is often set as “10”). You might then be able to avoid the
> need for MergeRecord at all and the conversion of JSON to Parquet could be
> done with a ConvertRecord processor instead, for example.
>
>
> Cheers,
>
> ---
> Chris Sampson
> IT Consultant
> chris.sampson@naimuri.com
>
>
> On 12 Mar 2024, at 08:38, edi mari <ed...@gmail.com> wrote:
>
> Hi,
> My task is to query Elastic and save the results in a Parquet file.
> I'm querying Elastic using the *PaginatedJsonQueryElasticsearch*
> processor.
> The files coming from Elastic are in JSON format, and I'm using the
> *MergeRecord* processor to convert the JSON to Parquet format and merge
> the result into one file .
> The *Record Reader* uses the *JsonTreeReader* and *Record Writer* uses
> the *ParquetRecordSetWriter* controller, which uses AVRO schema (Schema
> Text) to help with the conversion task.
>
> The process works fine, the only problem is that it takes too much time.
> Converting 5 MB takes 30 minutes.
>
> Do you have any idea how to enhance the process?
>
> <image.png>
>
> <image.png>
> <image.png>
>
>
>
Re: MergeRecord takes too much time
Posted by Chris Sampson <ch...@naimuri.com>.
I’m not an expert with MergeRecord, but looking at your screenshots, I’d guess that your setup is taking that long to reach one of the defined “maximum” settings, e.g. 2GB, 5,000,000 records, or 3600 seconds (1 hour).
How large (number of records and content size in bytes) are the typical FlowFiles you’re sending to Merge Record, and how often do they arrive? For example, if you’re getting results form Elasticsearch in chunks of 10,000 records per response and that’s 10MB in size every 1 second, it’s going to take a long time to meet any of the defined maximums.
How have you configured the Elasticsearch processor and what are you trying to combine together in your flow? Is it that you’re outputting a single FlowFile per Response (the default setting for “Search Results Split”), then trying to merge together all responses from a single query into one FlowFile? If so, I’d suggest changing the Elasticsearch processor’s “Search Results Split” to be “Per Query” instead, and increase the “Size” setting (leaving this blank will use the Elasticsearch default page size, which is often set as “10”). You might then be able to avoid the need for MergeRecord at all and the conversion of JSON to Parquet could be done with a ConvertRecord processor instead, for example.
Cheers,
---
Chris Sampson
IT Consultant
chris.sampson@naimuri.com
> On 12 Mar 2024, at 08:38, edi mari <ed...@gmail.com> wrote:
>
> Hi,
> My task is to query Elastic and save the results in a Parquet file.
> I'm querying Elastic using the PaginatedJsonQueryElasticsearch processor.
> The files coming from Elastic are in JSON format, and I'm using the MergeRecord processor to convert the JSON to Parquet format and merge the result into one file .
> The Record Reader uses the JsonTreeReader and Record Writer uses the ParquetRecordSetWriter controller, which uses AVRO schema (Schema Text) to help with the conversion task.
>
> The process works fine, the only problem is that it takes too much time.
> Converting 5 MB takes 30 minutes.
>
> Do you have any idea how to enhance the process?
>
> <image.png>
>
> <image.png>
> <image.png>
Re: MergeRecord takes too much time
Posted by Russell Bateman <ru...@windofkeltia.com>.
Atul,
Please look here for ending your subscription with the Apache NiFi Users
mailing list:
https://nifi.staged.apache.org/community/contact/
On 3/18/24 10:26, Atul Wankhade wrote:
> unsubscribe
>
> On Tue, Mar 12, 2024 at 8:39 AM edi mari <ed...@gmail.com> wrote:
>
> Hi,
> My task is to query Elastic and save the results in a Parquet file.
> I'm querying Elastic using the *PaginatedJsonQueryElasticsearch*
> processor.
> The files coming from Elastic are in JSON format, and I'm using
> the *MergeRecord* processor to convert the JSON to Parquet format
> and merge the result into one file .
> The *Record Reader* uses the *JsonTreeReader* and *Record Writer*
> uses the *ParquetRecordSetWriter* controller, which uses AVRO
> schema (Schema Text) to help with the conversion task.
>
> The process works fine, the only problem is that it takes too much
> time.
> Converting 5 MB takes 30 minutes.
>
> Do you have any idea how to enhance the process?
>
> image.png
>
> image.png
> image.png
>
Re: MergeRecord takes too much time
Posted by Atul Wankhade <at...@gmail.com>.
unsubscribe
On Tue, Mar 12, 2024 at 8:39 AM edi mari <ed...@gmail.com> wrote:
> Hi,
> My task is to query Elastic and save the results in a Parquet file.
> I'm querying Elastic using the *PaginatedJsonQueryElasticsearch*
> processor.
> The files coming from Elastic are in JSON format, and I'm using the
> *MergeRecord* processor to convert the JSON to Parquet format and merge
> the result into one file .
> The *Record Reader* uses the *JsonTreeReader* and *Record Writer* uses
> the *ParquetRecordSetWriter* controller, which uses AVRO schema (Schema
> Text) to help with the conversion task.
>
> The process works fine, the only problem is that it takes too much time.
> Converting 5 MB takes 30 minutes.
>
> Do you have any idea how to enhance the process?
>
> [image: image.png]
>
> [image: image.png]
> [image: image.png]
>