This is an automated email from the ASF dual-hosted git repository.
bamaer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 77e9a2296e Fixes #2929 : Stream Enhanced JSON Output to the file
(#8713)
77e9a2296e is described below
commit 77e9a2296e02f6dce72ba7bf83914d9d27822251
Author: Bart Maertens <[email protected]>
AuthorDate: Tue Oct 6 15:09:32 2026 +0200
Fixes #2929 : Stream Enhanced JSON Output to the file (#8713)
* Issue #2929 : Stream Enhanced JSON Output to the file
* Issue #2929 : Close the JSON file generator in dispose() so a failed run
keeps a well-formed file
---
.../pipeline/transforms/enhancedjsonoutput.adoc | 18 +-
.../json/0018-enhanced-json-output-streaming.hpl | 201 ++++++++++
.../0019-enhanced-json-output-grouped-file.hpl | 427 +++++++++++++++++++++
.../main-0018-enhanced-json-output-streaming.hwf | 188 +++++++++
...main-0019-enhanced-json-output-grouped-file.hwf | 273 +++++++++++++
.../transforms/jsonoutputenhanced/JsonEOutput.java | 338 +++++++++-------
.../jsonoutputenhanced/JsonEOutputData.java | 22 +-
7 files changed, 1332 insertions(+), 135 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/enhancedjsonoutput.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/enhancedjsonoutput.adoc
index a7d2e10aa0..38aa8b3e2a 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/enhancedjsonoutput.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/enhancedjsonoutput.adoc
@@ -43,9 +43,13 @@ General tab allows to specify the type of transform
operation (output JSON for t
Currently three types of operation are available:
1. Output value - only pass output JSON as a transform output field, do not
dump to output file
-2. Write to file - only write to file, do not pass to output field
+2. Write to file - only write to file, do not pass to output field. The input
rows are passed on to the next transform as they are.
3. Output value and write to file - dump to file and pass generated JSON as a
transform output field
+The file is written while the rows come in, so it does not have to fit in
memory.
+The output field does: without group keys, it holds the JSON of all rows.
+Use 'Write to file' for large files.
+
|JSON block name|If specified, the output of the transform will always be a
JSON object with a single first-level node, whose name will be this value.
If empty, the transform can output a JSON array or object, depending on the
settings in the other tabs.
@@ -69,7 +73,7 @@ Or if file does not exist, it will be created as in previous
case.
TIP: If you want to create a ND-JSON (Newline-Delimited JSON) file, you may
get better results by outputting the JSON rows and then printing them with a
xref:pipeline/transforms/textfileoutput.adoc[Text file output] transform
-|Split JSON after n rows|If this number N is larger than zero, split the
resulting JSON file into multiple parts of N rows.
+|Split JSON after n rows|If this number N is larger than zero, split the
resulting JSON file into multiple parts of N items: N rows, or N groups when
group keys are used.
|Create Parent folder|Check this option to create the folders structure, if
some of them are missing in the provided path.
If this option is not checked and the full path cannot be found, the transform
will fail.
|Do not open create at start|If not checked - file (and in some cases parent
folder) will be created/opened to write during pipeline initialization.
@@ -93,6 +97,16 @@ Rows with the same values in the key fields allow you to
generate JSON fragments
If no group field is defined, all the rows will be grouped in a JSON array and
the transform output will be a single row and a single column.
+In the output file, every group is one item: the key fields plus the JSON of
the group, under the name of the output value.
+For example, with key field `group` and output value `rows`:
+
+[source,json]
+----
+[{"group":"a","rows":[{"id":1},{"id":2}]},{"group":"b","rows":{"id":3}}]
+----
+
+A JSON block name wraps the whole file, not every group:
`{"data":[{"group":"a","rows":[...]}, ...]}`.
+
[options="header"]
|===
|Option|Description
diff --git a/integration-tests/json/0018-enhanced-json-output-streaming.hpl
b/integration-tests/json/0018-enhanced-json-output-streaming.hpl
new file mode 100644
index 0000000000..22ac42c597
--- /dev/null
+++ b/integration-tests/json/0018-enhanced-json-output-streaming.hpl
@@ -0,0 +1,201 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<pipeline>
+ <info>
+ <name>0018-enhanced-json-output-streaming</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ <parameter>
+ <name>OUTPUT_FILE</name>
+ <default_value/>
+ <description>The JSON file to write, without extension</description>
+ </parameter>
+ <parameter>
+ <name>ROWS</name>
+ <default_value>1000</default_value>
+ <description>The number of rows to write</description>
+ </parameter>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2026/09/30 16:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/09/30 16:00:00.000</modified_date>
+ </info>
+ <notepads>
+ <notepad>
+ <backgroundcolorblue>251</backgroundcolorblue>
+ <backgroundcolorgreen>232</backgroundcolorgreen>
+ <backgroundcolorred>201</backgroundcolorred>
+ <bordercolorblue>90</bordercolorblue>
+ <bordercolorgreen>58</bordercolorgreen>
+ <bordercolorred>14</bordercolorred>
+ <fontbold>N</fontbold>
+ <fontcolorblue>90</fontcolorblue>
+ <fontcolorgreen>58</fontcolorgreen>
+ <fontcolorred>14</fontcolorred>
+ <fontitalic>N</fontitalic>
+ <fontname>Noto Sans</fontname>
+ <fontsize>10</fontsize>
+ <height>60</height>
+ <xloc>64</xloc>
+ <yloc>144</yloc>
+ <note>Run by main-0018 in a separate hop-run with a small heap. The JSON
file is several
+times larger than that heap, so it only gets written when the transform
streams it.</note>
+ <width>560</width>
+ </notepad>
+ </notepads>
+ <order>
+ <hop>
+ <from>generate rows</from>
+ <to>add id</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>add id</from>
+ <to>Enhanced JSON Output</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>generate rows</name>
+ <type>RowGenerator</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <currency/>
+ <decimal/>
+ <format/>
+ <group/>
+ <length>-1</length>
+ <name>payload</name>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <type>String</type>
+
<nullif>xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx</nullif>
+ </field>
+ </fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <last_time_field>FiveSecondsAgo</last_time_field>
+ <limit>${ROWS}</limit>
+ <never_ending>N</never_ending>
+ <row_time_field>now</row_time_field>
+ <attributes/>
+ <GUI>
+ <xloc>64</xloc>
+ <yloc>64</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>add id</name>
+ <type>Sequence</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <counter_name/>
+ <use_counter>Y</use_counter>
+ <connection/>
+ <use_database>N</use_database>
+ <increment_by>1</increment_by>
+ <max_value>999999999</max_value>
+ <schema/>
+ <seqname>SEQ_</seqname>
+ <start_at>1</start_at>
+ <valuename>id</valuename>
+ <attributes/>
+ <GUI>
+ <xloc>240</xloc>
+ <yloc>64</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Enhanced JSON Output</name>
+ <type>EnhancedJsonOutput</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <outputValue>outputValue</outputValue>
+ <jsonBloc>data</jsonBloc>
+ <operation_type>writetofile</operation_type>
+ <use_arrays_with_single_instance>N</use_arrays_with_single_instance>
+ <use_single_item_per_group>N</use_single_item_per_group>
+ <json_prittified>N</json_prittified>
+ <encoding>UTF-8</encoding>
+ <addtoresult>N</addtoresult>
+ <file>
+ <name>${OUTPUT_FILE}</name>
+ <split_output_after>0</split_output_after>
+ <extention>json</extention>
+ <append>N</append>
+ <create_parent_folder>Y</create_parent_folder>
+ <doNotOpenNewFileInit>N</doNotOpenNewFileInit>
+ </file>
+ <json_size_field/>
+ <key_fields>
+ </key_fields>
+ <fields>
+ <field>
+ <name>id</name>
+ <element>id</element>
+ <json_fragment>N</json_fragment>
+ <is_without_enclosing>N</is_without_enclosing>
+ <remove_if_blank>N</remove_if_blank>
+ </field>
+ <field>
+ <name>payload</name>
+ <element>payload</element>
+ <json_fragment>N</json_fragment>
+ <is_without_enclosing>N</is_without_enclosing>
+ <remove_if_blank>N</remove_if_blank>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>416</xloc>
+ <yloc>64</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/json/0019-enhanced-json-output-grouped-file.hpl
b/integration-tests/json/0019-enhanced-json-output-grouped-file.hpl
new file mode 100644
index 0000000000..9a0fd3745a
--- /dev/null
+++ b/integration-tests/json/0019-enhanced-json-output-grouped-file.hpl
@@ -0,0 +1,427 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<pipeline>
+ <info>
+ <name>0019-enhanced-json-output-grouped-file</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2026/09/30 16:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/09/30 16:00:00.000</modified_date>
+ </info>
+ <notepads>
+ <notepad>
+ <backgroundcolorblue>251</backgroundcolorblue>
+ <backgroundcolorgreen>232</backgroundcolorgreen>
+ <backgroundcolorred>201</backgroundcolorred>
+ <bordercolorblue>90</bordercolorblue>
+ <bordercolorgreen>58</bordercolorgreen>
+ <bordercolorred>14</bordercolorred>
+ <fontbold>N</fontbold>
+ <fontcolorblue>90</fontcolorblue>
+ <fontcolorgreen>58</fontcolorgreen>
+ <fontcolorred>14</fontcolorred>
+ <fontitalic>N</fontitalic>
+ <fontname>Noto Sans</fontname>
+ <fontsize>10</fontsize>
+ <height>80</height>
+ <xloc>64</xloc>
+ <yloc>416</yloc>
+ <note>With group keys every group is one item in the file: the key
fields plus the group's JSON
+under the output value name. A JSON block name only wraps the file, not every
group.
+main-0019 checks the files and the rows.
+'write to file' only writes the file and passes the input rows on as they are.
+'split, output value and file' splits after 2 groups, and still outputs the
row of the last group.</note>
+ <width>620</width>
+ </notepad>
+ </notepads>
+ <order>
+ <hop>
+ <from>rows in 3 groups</from>
+ <to>write to file</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>write to file</from>
+ <to>passed rows</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>rows in 3 groups</from>
+ <to>split, output value and file</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>split, output value and file</from>
+ <to>JSON rows</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>rows in 3 groups</from>
+ <to>write to file with a JSON block</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>rows in 3 groups</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>N</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <set_empty_string>N</set_empty_string>
+ <length>-1</length>
+ <name>grp</name>
+ <precision>-1</precision>
+ <type>String</type>
+ </field>
+ <field>
+ <set_empty_string>N</set_empty_string>
+ <length>-1</length>
+ <name>id</name>
+ <precision>-1</precision>
+ <type>Integer</type>
+ </field>
+ <field>
+ <set_empty_string>N</set_empty_string>
+ <length>-1</length>
+ <name>name</name>
+ <precision>-1</precision>
+ <type>String</type>
+ </field>
+ </fields>
+ <data>
+ <line>
+ <item>a</item>
+ <item>1</item>
+ <item>x1</item>
+ </line>
+ <line>
+ <item>a</item>
+ <item>2</item>
+ <item>x2</item>
+ </line>
+ <line>
+ <item>b</item>
+ <item>3</item>
+ <item>x3</item>
+ </line>
+ <line>
+ <item>b</item>
+ <item>4</item>
+ <item/>
+ </line>
+ <line>
+ <item>c</item>
+ <item>5</item>
+ <item>x5</item>
+ </line>
+ </data>
+ <attributes/>
+ <GUI>
+ <xloc>64</xloc>
+ <yloc>128</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>write to file</name>
+ <type>EnhancedJsonOutput</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <outputValue>rows</outputValue>
+ <jsonBloc/>
+ <operation_type>writetofile</operation_type>
+ <use_arrays_with_single_instance>N</use_arrays_with_single_instance>
+ <use_single_item_per_group>N</use_single_item_per_group>
+ <json_prittified>N</json_prittified>
+ <encoding>UTF-8</encoding>
+ <addtoresult>N</addtoresult>
+ <file>
+
<name>${PROJECT_HOME}/output/0019-enhanced-json-output-grouped-file/grouped</name>
+ <split_output_after>0</split_output_after>
+ <extention>json</extention>
+ <append>N</append>
+ <create_parent_folder>Y</create_parent_folder>
+ <doNotOpenNewFileInit>N</doNotOpenNewFileInit>
+ </file>
+ <json_size_field/>
+ <key_fields>
+ <key_field>
+ <key_field_name>grp</key_field_name>
+ <key_field_element>group</key_field_element>
+ </key_field>
+ </key_fields>
+ <fields>
+ <field>
+ <name>id</name>
+ <element>id</element>
+ <json_fragment>N</json_fragment>
+ <is_without_enclosing>N</is_without_enclosing>
+ <remove_if_blank>N</remove_if_blank>
+ </field>
+ <field>
+ <name>name</name>
+ <element>name</element>
+ <json_fragment>N</json_fragment>
+ <is_without_enclosing>N</is_without_enclosing>
+ <remove_if_blank>Y</remove_if_blank>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>320</xloc>
+ <yloc>64</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>write to file with a JSON block</name>
+ <type>EnhancedJsonOutput</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <outputValue>rows</outputValue>
+ <jsonBloc>data</jsonBloc>
+ <operation_type>writetofile</operation_type>
+ <use_arrays_with_single_instance>N</use_arrays_with_single_instance>
+ <use_single_item_per_group>N</use_single_item_per_group>
+ <json_prittified>N</json_prittified>
+ <encoding>UTF-8</encoding>
+ <addtoresult>N</addtoresult>
+ <file>
+
<name>${PROJECT_HOME}/output/0019-enhanced-json-output-grouped-file/grouped-block</name>
+ <split_output_after>0</split_output_after>
+ <extention>json</extention>
+ <append>N</append>
+ <create_parent_folder>Y</create_parent_folder>
+ <doNotOpenNewFileInit>N</doNotOpenNewFileInit>
+ </file>
+ <json_size_field/>
+ <key_fields>
+ <key_field>
+ <key_field_name>grp</key_field_name>
+ <key_field_element>group</key_field_element>
+ </key_field>
+ </key_fields>
+ <fields>
+ <field>
+ <name>id</name>
+ <element>id</element>
+ <json_fragment>N</json_fragment>
+ <is_without_enclosing>N</is_without_enclosing>
+ <remove_if_blank>N</remove_if_blank>
+ </field>
+ <field>
+ <name>name</name>
+ <element>name</element>
+ <json_fragment>N</json_fragment>
+ <is_without_enclosing>N</is_without_enclosing>
+ <remove_if_blank>Y</remove_if_blank>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>320</xloc>
+ <yloc>320</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>passed rows</name>
+ <type>TextFileOutput</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <separator>|</separator>
+ <enclosure/>
+ <enclosure_forced>N</enclosure_forced>
+ <enclosure_fix_disabled>N</enclosure_fix_disabled>
+ <header>N</header>
+ <footer>N</footer>
+ <format>UNIX</format>
+ <compression>None</compression>
+ <encoding>UTF-8</encoding>
+ <endedLine/>
+ <fileNameInField>N</fileNameInField>
+ <fileNameField/>
+ <create_parent_folder>Y</create_parent_folder>
+ <file>
+
<name>${PROJECT_HOME}/output/0019-enhanced-json-output-grouped-file/passed-rows</name>
+ <servlet_output>N</servlet_output>
+ <do_not_open_new_file_init>N</do_not_open_new_file_init>
+ <extention>txt</extention>
+ <append>N</append>
+ <split>N</split>
+ <haspartno>N</haspartno>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <SpecifyFormat>N</SpecifyFormat>
+ <date_time_format/>
+ <add_to_result_filenames>N</add_to_result_filenames>
+ <pad>N</pad>
+ <fast_dump>N</fast_dump>
+ <splitevery/>
+ </file>
+ <fields>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>560</xloc>
+ <yloc>64</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>split, output value and file</name>
+ <type>EnhancedJsonOutput</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <outputValue>rows</outputValue>
+ <jsonBloc/>
+ <operation_type>both</operation_type>
+ <use_arrays_with_single_instance>N</use_arrays_with_single_instance>
+ <use_single_item_per_group>N</use_single_item_per_group>
+ <json_prittified>N</json_prittified>
+ <encoding>UTF-8</encoding>
+ <addtoresult>N</addtoresult>
+ <file>
+
<name>${PROJECT_HOME}/output/0019-enhanced-json-output-grouped-file/split</name>
+ <split_output_after>2</split_output_after>
+ <extention>json</extention>
+ <append>N</append>
+ <create_parent_folder>Y</create_parent_folder>
+ <doNotOpenNewFileInit>N</doNotOpenNewFileInit>
+ </file>
+ <json_size_field/>
+ <key_fields>
+ <key_field>
+ <key_field_name>grp</key_field_name>
+ <key_field_element>group</key_field_element>
+ </key_field>
+ </key_fields>
+ <fields>
+ <field>
+ <name>id</name>
+ <element>id</element>
+ <json_fragment>N</json_fragment>
+ <is_without_enclosing>N</is_without_enclosing>
+ <remove_if_blank>N</remove_if_blank>
+ </field>
+ <field>
+ <name>name</name>
+ <element>name</element>
+ <json_fragment>N</json_fragment>
+ <is_without_enclosing>N</is_without_enclosing>
+ <remove_if_blank>Y</remove_if_blank>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>320</xloc>
+ <yloc>192</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>JSON rows</name>
+ <type>TextFileOutput</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <separator>|</separator>
+ <enclosure/>
+ <enclosure_forced>N</enclosure_forced>
+ <enclosure_fix_disabled>N</enclosure_fix_disabled>
+ <header>N</header>
+ <footer>N</footer>
+ <format>UNIX</format>
+ <compression>None</compression>
+ <encoding>UTF-8</encoding>
+ <endedLine/>
+ <fileNameInField>N</fileNameInField>
+ <fileNameField/>
+ <create_parent_folder>Y</create_parent_folder>
+ <file>
+
<name>${PROJECT_HOME}/output/0019-enhanced-json-output-grouped-file/split-rows</name>
+ <servlet_output>N</servlet_output>
+ <do_not_open_new_file_init>N</do_not_open_new_file_init>
+ <extention>txt</extention>
+ <append>N</append>
+ <split>N</split>
+ <haspartno>N</haspartno>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <SpecifyFormat>N</SpecifyFormat>
+ <date_time_format/>
+ <add_to_result_filenames>N</add_to_result_filenames>
+ <pad>N</pad>
+ <fast_dump>N</fast_dump>
+ <splitevery/>
+ </file>
+ <fields>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>560</xloc>
+ <yloc>192</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git
a/integration-tests/json/main-0018-enhanced-json-output-streaming.hwf
b/integration-tests/json/main-0018-enhanced-json-output-streaming.hwf
new file mode 100644
index 0000000000..c4aa06758e
--- /dev/null
+++ b/integration-tests/json/main-0018-enhanced-json-output-streaming.hwf
@@ -0,0 +1,188 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<workflow>
+ <name>main-0018-enhanced-json-output-streaming</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Enhanced JSON Output writes a file that is larger than the
heap</description>
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2026/09/30 16:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/09/30 16:00:00.000</modified_date>
+ <parameters>
+ </parameters>
+ <actions>
+ <action>
+ <name>Start</name>
+ <description/>
+ <type>SPECIAL</type>
+ <attributes/>
+ <DayOfMonth>1</DayOfMonth>
+ <hour>12</hour>
+ <intervalMinutes>60</intervalMinutes>
+ <intervalSeconds>0</intervalSeconds>
+ <minutes>0</minutes>
+ <repeat>N</repeat>
+ <schedulerType>0</schedulerType>
+ <weekDay>1</weekDay>
+ <parallel>N</parallel>
+ <xloc>112</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Write a JSON file larger than the heap</name>
+ <description/>
+ <type>SHELL</type>
+ <attributes/>
+ <filename/>
+ <work_directory/>
+ <arg_from_previous>N</arg_from_previous>
+ <exec_per_row>N</exec_per_row>
+ <set_logfile>N</set_logfile>
+ <logfile/>
+ <set_append_logfile>N</set_append_logfile>
+ <logext/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <insertScript>Y</insertScript>
+ <script>OUT="${PROJECT_HOME}/output/0018-enhanced-json-output-streaming"
+LOG="${OUT}.log"
+FILE="${OUT}/streamed.json"
+
+rm -rf "$OUT" "$LOG"
+mkdir -p "$OUT"
+
+# One million rows make a JSON file of about 127MB. The child hop-run gets a
128MB heap, a good
+# part of which Hop needs for itself. Building the file in memory runs out of
heap, streaming it
+# does not. Without ExitOnOutOfMemoryError the child keeps hanging after
running out of memory.
+if ! HOP_OPTIONS="-Xmx128m -XX:+ExitOnOutOfMemoryError" timeout 600 sh
hop-run.sh -r 'local' \
+ -f "${PROJECT_HOME}/0018-enhanced-json-output-streaming.hpl" \
+ -p ROWS=1000000 -p OUTPUT_FILE="${OUT}/streamed" -l Basic > "$LOG"
2>&1; then
+ echo "Writing the JSON file with a 128MB heap failed:"
+ tail -n 50 "$LOG"
+ exit 1
+fi
+
+PAYLOAD="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+FIRST="{\"data\":[{\"id\":1,\"payload\":\"${PAYLOAD}\"},"
+LAST="{\"id\":1000000,\"payload\":\"${PAYLOAD}\"}]}"
+
+failures=0
+fail() {
+ echo "FAIL: $1"
+ failures=$((failures + 1))
+}
+
+if [ ! -f "$FILE" ]; then
+ fail "${FILE} was not written"
+else
+ [ "$(head -c ${#FIRST} "$FILE")" = "$FIRST" ] || fail "the file does not
start with ${FIRST}"
+ [ "$(tail -c ${#LAST} "$FILE")" = "$LAST" ] || fail "the file does not end
with ${LAST}"
+ # One object per row plus the one around the JSON block
+ objects="$(tr -cd '{' < "$FILE" | wc -c | tr -d ' ')"
+ [ "$objects" = "1000001" ] || fail "expected 1000001 JSON objects, found
${objects}"
+ size="$(wc -c < "$FILE" | tr -d ' ')"
+ [ "$size" = "126888906" ] || fail "expected 126888906 bytes, found ${size}"
+fi
+
+# Do not leave 127MB behind
+rm -f "$FILE"
+
+[ "$failures" -eq 0 ]</script>
+ <loglevel>Basic</loglevel>
+ <parallel>N</parallel>
+ <xloc>320</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Success</name>
+ <description/>
+ <type>SUCCESS</type>
+ <attributes/>
+ <parallel>N</parallel>
+ <xloc>560</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Failure: the JSON file was not streamed</name>
+ <description/>
+ <type>ABORT</type>
+ <attributes/>
+ <always_log_rows>N</always_log_rows>
+ <message>Failure: the JSON file was not streamed</message>
+ <parallel>N</parallel>
+ <xloc>560</xloc>
+ <yloc>192</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>Write a JSON file larger than the heap</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ <hop>
+ <from>Write a JSON file larger than the heap</from>
+ <to>Success</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>Write a JSON file larger than the heap</from>
+ <to>Failure: the JSON file was not streamed</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ </hops>
+ <notepads>
+ <notepad>
+ <backgroundcolorblue>251</backgroundcolorblue>
+ <backgroundcolorgreen>232</backgroundcolorgreen>
+ <backgroundcolorred>201</backgroundcolorred>
+ <bordercolorblue>90</bordercolorblue>
+ <bordercolorgreen>58</bordercolorgreen>
+ <bordercolorred>14</bordercolorred>
+ <fontbold>N</fontbold>
+ <fontcolorblue>90</fontcolorblue>
+ <fontcolorgreen>58</fontcolorgreen>
+ <fontcolorred>14</fontcolorred>
+ <fontitalic>N</fontitalic>
+ <fontname>Noto Sans</fontname>
+ <fontsize>10</fontsize>
+ <height>80</height>
+ <xloc>112</xloc>
+ <yloc>256</yloc>
+ <note>Issue #2929: Enhanced JSON Output built the whole file in memory
and ran out of heap on
+large files. 0018-enhanced-json-output-streaming.hpl runs in a separate
hop-run with a 128MB heap
+and writes a JSON file of about 127MB. That only works when the rows are
streamed to the file.</note>
+ <width>620</width>
+ </notepad>
+ </notepads>
+ <attributes/>
+</workflow>
diff --git
a/integration-tests/json/main-0019-enhanced-json-output-grouped-file.hwf
b/integration-tests/json/main-0019-enhanced-json-output-grouped-file.hwf
new file mode 100644
index 0000000000..bb83135930
--- /dev/null
+++ b/integration-tests/json/main-0019-enhanced-json-output-grouped-file.hwf
@@ -0,0 +1,273 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<workflow>
+ <name>main-0019-enhanced-json-output-grouped-file</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Enhanced JSON Output writes every group to the
file</description>
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2026/09/30 16:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/09/30 16:00:00.000</modified_date>
+ <parameters>
+ </parameters>
+ <actions>
+ <action>
+ <name>Start</name>
+ <description/>
+ <type>SPECIAL</type>
+ <attributes/>
+ <DayOfMonth>1</DayOfMonth>
+ <hour>12</hour>
+ <intervalMinutes>60</intervalMinutes>
+ <intervalSeconds>0</intervalSeconds>
+ <minutes>0</minutes>
+ <repeat>N</repeat>
+ <schedulerType>0</schedulerType>
+ <weekDay>1</weekDay>
+ <parallel>N</parallel>
+ <xloc>112</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Remove old output</name>
+ <description/>
+ <type>SHELL</type>
+ <attributes/>
+ <filename/>
+ <work_directory/>
+ <arg_from_previous>N</arg_from_previous>
+ <exec_per_row>N</exec_per_row>
+ <set_logfile>N</set_logfile>
+ <logfile/>
+ <set_append_logfile>N</set_append_logfile>
+ <logext/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <insertScript>Y</insertScript>
+ <script>rm -rf
"${PROJECT_HOME}/output/0019-enhanced-json-output-grouped-file"</script>
+ <loglevel>Basic</loglevel>
+ <parallel>N</parallel>
+ <xloc>272</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>0019-enhanced-json-output-grouped-file</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <clear_files>N</clear_files>
+ <clear_rows>N</clear_rows>
+ <create_parent_folder>N</create_parent_folder>
+ <exec_per_row>N</exec_per_row>
+
<filename>${PROJECT_HOME}/0019-enhanced-json-output-grouped-file.hpl</filename>
+ <loglevel>Basic</loglevel>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <params_from_previous>N</params_from_previous>
+ <run_configuration>local</run_configuration>
+ <set_append_logfile>N</set_append_logfile>
+ <set_logfile>N</set_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <parallel>N</parallel>
+ <xloc>464</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Check the files and rows</name>
+ <description/>
+ <type>SHELL</type>
+ <attributes/>
+ <filename/>
+ <work_directory/>
+ <arg_from_previous>N</arg_from_previous>
+ <exec_per_row>N</exec_per_row>
+ <set_logfile>N</set_logfile>
+ <logfile/>
+ <set_append_logfile>N</set_append_logfile>
+ <logext/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <insertScript>Y</insertScript>
+
<script>OUT="${PROJECT_HOME}/output/0019-enhanced-json-output-grouped-file"
+
+failures=0
+# check <what> <file> <expected content>
+check() {
+ if [ ! -f "$2" ]; then
+ echo "FAIL: $1: $2 was not written"
+ failures=$((failures + 1))
+ return
+ fi
+ actual="$(cat "$2")"
+ if [ "$actual" != "$3" ]; then
+ echo "FAIL: $1"
+ echo " expected: $3"
+ echo " actual: $actual"
+ failures=$((failures + 1))
+ fi
+}
+
+check "every group is an item in the file" "$OUT/grouped.json" \
+
'[{"group":"a","rows":[{"id":1,"name":"x1"},{"id":2,"name":"x2"}]},{"group":"b","rows":[{"id":3,"name":"x3"},{"id":4}]},{"group":"c","rows":{"id":5,"name":"x5"}}]'
+
+check "a JSON block name only wraps the file" "$OUT/grouped-block.json" \
+
'{"data":[{"group":"a","rows":[{"id":1,"name":"x1"},{"id":2,"name":"x2"}]},{"group":"b","rows":[{"id":3,"name":"x3"},{"id":4}]},{"group":"c","rows":{"id":5,"name":"x5"}}]}'
+
+check "write to file passes the input rows on" "$OUT/passed-rows.txt" 'a|1|x1
+a|2|x2
+b|3|x3
+b|4|
+c|5|x5'
+
+check "the first split file holds 2 groups" "$OUT/split_0.json" \
+
'[{"group":"a","rows":[{"id":1,"name":"x1"},{"id":2,"name":"x2"}]},{"group":"b","rows":[{"id":3,"name":"x3"},{"id":4}]}]'
+
+check "the second split file holds the last group" "$OUT/split_1.json" \
+ '{"group":"c","rows":{"id":5,"name":"x5"}}'
+
+if [ -f "$OUT/split_2.json" ]; then
+ echo "FAIL: a third split file was written"
+ failures=$((failures + 1))
+fi
+
+check "splitting the file still outputs a row for every group"
"$OUT/split-rows.txt" 'a|[{"id":1,"name":"x1"},{"id":2,"name":"x2"}]
+b|[{"id":3,"name":"x3"},{"id":4}]
+c|{"id":5,"name":"x5"}'
+
+[ "$failures" -eq 0 ]</script>
+ <loglevel>Basic</loglevel>
+ <parallel>N</parallel>
+ <xloc>688</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Success</name>
+ <description/>
+ <type>SUCCESS</type>
+ <attributes/>
+ <parallel>N</parallel>
+ <xloc>896</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Failure: the pipeline failed</name>
+ <description/>
+ <type>ABORT</type>
+ <attributes/>
+ <always_log_rows>N</always_log_rows>
+ <message>Failure: the pipeline failed</message>
+ <parallel>N</parallel>
+ <xloc>464</xloc>
+ <yloc>208</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Failure: wrong files or rows</name>
+ <description/>
+ <type>ABORT</type>
+ <attributes/>
+ <always_log_rows>N</always_log_rows>
+ <message>Failure: wrong files or rows</message>
+ <parallel>N</parallel>
+ <xloc>688</xloc>
+ <yloc>208</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>Remove old output</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ <hop>
+ <from>Remove old output</from>
+ <to>0019-enhanced-json-output-grouped-file</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ <hop>
+ <from>0019-enhanced-json-output-grouped-file</from>
+ <to>Check the files and rows</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>0019-enhanced-json-output-grouped-file</from>
+ <to>Failure: the pipeline failed</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>Check the files and rows</from>
+ <to>Success</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>Check the files and rows</from>
+ <to>Failure: wrong files or rows</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ </hops>
+ <notepads>
+ <notepad>
+ <backgroundcolorblue>251</backgroundcolorblue>
+ <backgroundcolorgreen>232</backgroundcolorgreen>
+ <backgroundcolorred>201</backgroundcolorred>
+ <bordercolorblue>90</bordercolorblue>
+ <bordercolorgreen>58</bordercolorgreen>
+ <bordercolorred>14</bordercolorred>
+ <fontbold>N</fontbold>
+ <fontcolorblue>90</fontcolorblue>
+ <fontcolorgreen>58</fontcolorgreen>
+ <fontcolorred>14</fontcolorred>
+ <fontitalic>N</fontitalic>
+ <fontname>Noto Sans</fontname>
+ <fontsize>10</fontsize>
+ <height>80</height>
+ <xloc>112</xloc>
+ <yloc>288</yloc>
+ <note>Issue #2929: with group keys, Enhanced JSON Output only wrote the
last group to the file,
+and split files mixed rows with groups and dropped the output row of the last
group.
+'Write to file' sent the JSON downstream in a row that did not match its
fields.</note>
+ <width>620</width>
+ </notepad>
+ </notepads>
+ <attributes/>
+</workflow>
diff --git
a/plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutput.java
b/plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutput.java
index 31d29c9ae0..fb9035e157 100644
---
a/plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutput.java
+++
b/plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutput.java
@@ -17,8 +17,11 @@
package org.apache.hop.pipeline.transforms.jsonoutputenhanced;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.util.DefaultPrettyPrinter;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -60,6 +63,7 @@ public class JsonEOutput extends
BaseTransform<JsonEOutputMeta, JsonEOutputData>
public Object[] prevRow;
private JsonNodeFactory nc;
private ObjectMapper mapper;
+ private ObjectMapper fileMapper;
private ObjectNode currentNode;
public JsonEOutput(
@@ -83,14 +87,18 @@ public class JsonEOutput extends
BaseTransform<JsonEOutputMeta, JsonEOutputData>
setErrors(1);
return false;
}
- if (meta.getOperationType() ==
JsonEOutputMeta.OperationType.WRITE_TO_FILE
- || meta.getOperationType() == JsonEOutputMeta.OperationType.BOTH) {
- // Init global json items array only if output to file is needed
- data.jsonItems = new ArrayList<>();
- data.isWriteToFile = true;
- if (!meta.getFileSettings().isDoNotOpenNewFileInit()
- && data.isWriteToFile
- && !openNewFile()) {
+ data.isOutputValue = meta.getOperationType() !=
JsonEOutputMeta.OperationType.WRITE_TO_FILE;
+ data.isWriteToFile =
+ meta.getOperationType() ==
JsonEOutputMeta.OperationType.WRITE_TO_FILE
+ || meta.getOperationType() == JsonEOutputMeta.OperationType.BOTH;
+ // Without group keys every row is an item in the file, unless all rows
are merged into a
+ // single item. Group keys make every group an item.
+ data.streamFileRows =
+ data.isWriteToFile && meta.getKeyFields().isEmpty() &&
!meta.isUseSingleItemPerGroup();
+ data.collectGroupItems = data.isOutputValue || (data.isWriteToFile &&
!data.streamFileRows);
+
+ if (data.isWriteToFile) {
+ if (!meta.getFileSettings().isDoNotOpenNewFileInit() &&
!openNewFile()) {
logError(BaseMessages.getString(PKG, "JsonOutput.Error.OpenNewFile",
buildFilename()));
stopAll();
setErrors(1);
@@ -110,26 +118,13 @@ public class JsonEOutput extends
BaseTransform<JsonEOutputMeta, JsonEOutputData>
// This also waits for a row to be finished.
Object[] r = getRow();
if (r == null) {
- // only attempt writing to file when the first row is not empty
- if (data.isWriteToFile && !first &&
meta.getFileSettings().getSplitOutputAfter() == 0) {
- // no more input to be expected...
- // Let's output the remaining unsafe data
- outputRow(prevRow);
- writeJsonFile();
- setOutputDone();
- return false;
+ // no more input to be expected: finish the last group and the file
+ if (!first) {
+ finishGroup(prevRow);
}
-
- // Process the leftover data only when a split file size is defined
- // and there are still items pending.
- if (meta.getFileSettings().getSplitOutputAfter() > 0 &&
!data.jsonItems.isEmpty()) {
- serializeJson(data.jsonItems);
- writeJsonFile();
- setOutputDone();
- return false;
+ if (data.isWriteToFile) {
+ finishFile();
}
-
- outputRow(prevRow);
setOutputDone();
return false;
}
@@ -140,6 +135,11 @@ public class JsonEOutput extends
BaseTransform<JsonEOutputMeta, JsonEOutputData>
data.rowsAreSafe = false;
manageRowItems(r);
+
+ if (!data.isOutputValue) {
+ // The JSON only goes to the file: pass the rows on as they are
+ putRow(data.inputRowMeta, r);
+ }
return true;
}
@@ -164,13 +164,8 @@ public class JsonEOutput extends
BaseTransform<JsonEOutputMeta, JsonEOutputData>
itemNode = new ObjectNode(nc);
}
- if (!sameGroup && !data.jsonKeyGroupItems.isEmpty()) {
- // Output the new row
- if (isDebug()) {
- logDebug("Record Num: " + data.nrRow + " - Generating JSON chunk");
- }
- outputRow(prevRow);
- data.jsonKeyGroupItems = new ArrayList<>();
+ if (!sameGroup) {
+ finishGroup(prevRow);
}
for (int i = 0; i < data.nrFields; i++) {
@@ -283,32 +278,22 @@ public class JsonEOutput extends
BaseTransform<JsonEOutputMeta, JsonEOutputData>
break;
}
}
- if (meta.getFileSettings().getSplitOutputAfter() > 0) {
- data.jsonItems.add(itemNode);
- }
-
/*
* Only add a new item node if each row should produce a single JSON
object or in case of a
* single JSON object for a group of rows, if no item node was added yet.
This happens for the
* first new row of a group only.
*/
- if (!meta.isUseSingleItemPerGroup() || data.jsonKeyGroupItems.isEmpty()) {
+ if (data.collectGroupItems
+ && (!meta.isUseSingleItemPerGroup() ||
data.jsonKeyGroupItems.isEmpty())) {
data.jsonKeyGroupItems.add(itemNode);
}
+ if (data.streamFileRows) {
+ writeFileItem(itemNode);
+ }
+
prevRow = data.inputRowMeta.cloneRow(row); // copy the row to previous
data.nrRow++;
-
- if (meta.getFileSettings().getSplitOutputAfter() > 0
- && (data.nrRow) % meta.getFileSettings().getSplitOutputAfter() == 0) {
- // Output the new row
- if (isDebug()) {
- logDebug("Record Num: " + data.nrRow + " - Generating JSON chunk");
- }
- serializeJson(data.jsonItems);
- writeJsonFile();
- data.jsonItems = new ArrayList<>();
- }
}
private String getJsonAttributeName(JsonEOutputField field) {
@@ -321,113 +306,187 @@ public class JsonEOutput extends
BaseTransform<JsonEOutputMeta, JsonEOutputData>
return Const.NVL(elementName, field.getFieldName());
}
+ /**
+ * A group is complete: send its row to the output field and, when group
keys are used, write its
+ * item to the file.
+ */
+ private void finishGroup(Object[] groupRow) throws HopException {
+ if (Utils.isEmpty(data.jsonKeyGroupItems)) {
+ return;
+ }
+ if (isDebug()) {
+ logDebug("Record Num: " + data.nrRow + " - Generating JSON chunk");
+ }
+ if (data.isOutputValue) {
+ outputRow(groupRow);
+ }
+ if (data.isWriteToFile && !data.streamFileRows) {
+ if (meta.getKeyFields().isEmpty()) {
+ // All rows are merged into a single item
+ for (ObjectNode item : data.jsonKeyGroupItems) {
+ writeFileItem(item);
+ }
+ } else {
+ writeFileItem(buildGroupFileItem(groupRow));
+ }
+ }
+ data.jsonKeyGroupItems = new ArrayList<>();
+ }
+
private void outputRow(Object[] rowData) throws HopException {
- // We can now output an object
- ObjectNode globalItemNode = null;
+ serializeJson(data.jsonKeyGroupItems);
+ data.jsonLength = data.jsonSerialized.length();
- if (Utils.isEmpty(data.jsonKeyGroupItems)) return;
+ Object[] keyRow = getKeyValues(rowData);
- if (!data.jsonKeyGroupItems.isEmpty()) {
- serializeJson(data.jsonKeyGroupItems);
- }
+ Object[] additionalRowFields = new Object[2];
- data.jsonLength = data.jsonSerialized.length();
+ additionalRowFields[0] = data.jsonSerialized;
- if (data.outputRowMeta != null) {
+ // Fill accessory fields
+ if (!Utils.isEmpty(meta.getJsonSizeFieldName())) {
+ additionalRowFields[1] = data.jsonLength;
+ }
- Object[] keyRow = new Object[meta.getKeyFields().size()];
+ Object[] outputRowData = RowDataUtil.addRowData(keyRow, keyRow.length,
additionalRowFields);
+ incrementLinesOutput();
- // Create a new object with specified fields
- if (data.isWriteToFile) {
- globalItemNode = new ObjectNode(nc);
- }
+ putRow(data.outputRowMeta, outputRowData);
- for (int i = 0; i < meta.getKeyFields().size(); i++) {
- JsonEOutputKeyField keyField = meta.getKeyFields().get(i);
- try {
- IValueMeta vmi =
data.inputRowMeta.getValueMeta(data.keysGroupIndexes[i]);
- switch (vmi.getType()) {
- case IValueMeta.TYPE_BOOLEAN:
- keyRow[i] = data.inputRowMeta.getBoolean(rowData,
data.keysGroupIndexes[i]);
- if (data.isWriteToFile) {
- globalItemNode.put(getKeyJsonAttributeName(keyField),
(Boolean) keyRow[i]);
- }
- break;
- case IValueMeta.TYPE_INTEGER:
- keyRow[i] = data.inputRowMeta.getInteger(rowData,
data.keysGroupIndexes[i]);
- if (data.isWriteToFile) {
- globalItemNode.put(getKeyJsonAttributeName(keyField), (Long)
keyRow[i]);
- }
- break;
- case IValueMeta.TYPE_NUMBER:
- keyRow[i] = data.inputRowMeta.getNumber(rowData,
data.keysGroupIndexes[i]);
- if (data.isWriteToFile) {
- globalItemNode.put(getKeyJsonAttributeName(keyField), (Double)
keyRow[i]);
- }
- break;
- case IValueMeta.TYPE_BIGNUMBER:
- keyRow[i] = data.inputRowMeta.getBigNumber(rowData,
data.keysGroupIndexes[i]);
- if (data.isWriteToFile) {
- globalItemNode.put(getKeyJsonAttributeName(keyField),
(BigDecimal) keyRow[i]);
- }
- break;
- default:
- keyRow[i] = data.inputRowMeta.getString(rowData,
data.keysGroupIndexes[i]);
- if (data.isWriteToFile) {
- globalItemNode.put(getKeyJsonAttributeName(keyField), (String)
keyRow[i]);
- }
- break;
- }
- } catch (HopValueException e) {
- throw new HopException(
- "Error getting json values for key field: " +
keyField.getFieldName(), e);
- }
- }
+ // Data are safe
+ data.rowsAreSafe = true;
+ }
- if (data.isWriteToFile) {
- try {
- // JSON serialization here...
- JsonNode jsonNode = mapper.readTree(data.jsonSerialized);
- if (meta.getOutputValue() != null) {
- globalItemNode.set(meta.getOutputValue(), jsonNode);
- }
- } catch (IOException e) {
- throw new HopException("Error serializing JSON values", e);
- }
- data.jsonItems.add(globalItemNode);
+ private Object[] getKeyValues(Object[] rowData) throws HopException {
+ Object[] keyRow = new Object[meta.getKeyFields().size()];
+ for (int i = 0; i < meta.getKeyFields().size(); i++) {
+ JsonEOutputKeyField keyField = meta.getKeyFields().get(i);
+ try {
+ IValueMeta vmi =
data.inputRowMeta.getValueMeta(data.keysGroupIndexes[i]);
+ keyRow[i] =
+ switch (vmi.getType()) {
+ case IValueMeta.TYPE_BOOLEAN ->
+ data.inputRowMeta.getBoolean(rowData,
data.keysGroupIndexes[i]);
+ case IValueMeta.TYPE_INTEGER ->
+ data.inputRowMeta.getInteger(rowData,
data.keysGroupIndexes[i]);
+ case IValueMeta.TYPE_NUMBER ->
+ data.inputRowMeta.getNumber(rowData,
data.keysGroupIndexes[i]);
+ case IValueMeta.TYPE_BIGNUMBER ->
+ data.inputRowMeta.getBigNumber(rowData,
data.keysGroupIndexes[i]);
+ default -> data.inputRowMeta.getString(rowData,
data.keysGroupIndexes[i]);
+ };
+ } catch (HopValueException e) {
+ throw new HopException(
+ "Error getting json values for key field: " +
keyField.getFieldName(), e);
}
+ }
+ return keyRow;
+ }
- Object[] additionalRowFields = new Object[2];
+ /**
+ * The file item of a group: the key fields plus the group's items under the
output value name.
+ * The JSON block name only wraps the file, not every group.
+ */
+ private ObjectNode buildGroupFileItem(Object[] groupRow) throws HopException
{
+ ObjectNode groupItem = new ObjectNode(nc);
+ Object[] keyRow = getKeyValues(groupRow);
+ for (int i = 0; i < keyRow.length; i++) {
+ String name = getKeyJsonAttributeName(meta.getKeyFields().get(i));
+ switch (keyRow[i]) {
+ case null -> groupItem.putNull(name);
+ case Boolean b -> groupItem.put(name, b);
+ case Long l -> groupItem.put(name, l);
+ case Double d -> groupItem.put(name, d);
+ case BigDecimal bd -> groupItem.put(name, bd);
+ default -> groupItem.put(name, keyRow[i].toString());
+ }
+ }
+ groupItem.set(meta.getOutputValue(),
buildGroupValue(data.jsonKeyGroupItems));
+ return groupItem;
+ }
- additionalRowFields[0] = data.jsonSerialized;
+ /** The items of a group: an array, or the single item unless arrays are
forced. */
+ private JsonNode buildGroupValue(List<ObjectNode> items) {
+ if (items.size() > 1 || meta.isUseArrayWithSingleInstance()) {
+ return new ArrayNode(nc).addAll(items);
+ }
+ return items.get(0);
+ }
- // Fill accessory fields
- if (!Utils.isEmpty(meta.getJsonSizeFieldName())) {
- additionalRowFields[1] = data.jsonLength;
+ /**
+ * Write an item to the file straight away, so the file never has to fit in
memory. The first item
+ * is held back: a file with a single item holds that item, not an array,
unless arrays are
+ * forced.
+ */
+ private void writeFileItem(JsonNode item) throws HopException {
+ try {
+ if (data.fileItemCount == 0) {
+ data.pendingFileItem = item;
+ } else {
+ if (data.fileItemCount == 1) {
+ startFileDocument(true);
+ data.fileGenerator.writeTree(data.pendingFileItem);
+ data.pendingFileItem = null;
+ }
+ data.fileGenerator.writeTree(item);
}
-
- Object[] outputRowData = RowDataUtil.addRowData(keyRow, keyRow.length,
additionalRowFields);
+ } catch (IOException e) {
+ throw new HopTransformException(BaseMessages.getString(PKG,
"JsonOutput.Error.Writing"), e);
+ }
+ data.fileItemCount++;
+ if (!data.isOutputValue) {
incrementLinesOutput();
-
- putRow(data.outputRowMeta, outputRowData);
}
- // Data are safe
- data.rowsAreSafe = true;
+ int splitOutputAfter = meta.getFileSettings().getSplitOutputAfter();
+ if (splitOutputAfter > 0 && data.fileItemCount >= splitOutputAfter) {
+ finishFile();
+ }
}
- private void writeJsonFile() throws HopTransformException {
- // Open a file
- if (data.isWriteToFile && !openNewFile())
+ private void startFileDocument(boolean array) throws IOException,
HopTransformException {
+ if (!openNewFile()) {
throw new HopTransformException(
BaseMessages.getString(PKG, "JsonOutput.Error.OpenNewFile",
buildFilename()));
- // Write data to file
+ }
+ data.fileGenerator = fileMapper.getFactory().createGenerator(data.writer);
+ // The file is closed separately, with its lineage
+ data.fileGenerator.disable(JsonGenerator.Feature.AUTO_CLOSE_TARGET);
+ if (meta.isJsonPrettified()) {
+ data.fileGenerator.setPrettyPrinter(new DefaultPrettyPrinter());
+ }
+ if (!Utils.isEmpty(meta.getJsonBloc())) {
+ data.fileGenerator.writeStartObject();
+ data.fileGenerator.writeFieldName(meta.getJsonBloc());
+ }
+ if (array) {
+ data.fileGenerator.writeStartArray();
+ }
+ }
+
+ /** Close the JSON document and the file, if any item was written to it. */
+ private void finishFile() throws HopTransformException {
+ if (data.fileItemCount == 0) {
+ return;
+ }
try {
- data.writer.write(data.jsonSerialized);
- } catch (Exception e) {
+ if (data.fileItemCount == 1) {
+ startFileDocument(meta.isUseArrayWithSingleInstance());
+ data.fileGenerator.writeTree(data.pendingFileItem);
+ }
+ if (data.fileItemCount > 1 || meta.isUseArrayWithSingleInstance()) {
+ data.fileGenerator.writeEndArray();
+ }
+ if (!Utils.isEmpty(meta.getJsonBloc())) {
+ data.fileGenerator.writeEndObject();
+ }
+ data.fileGenerator.close();
+ } catch (IOException e) {
throw new HopTransformException(BaseMessages.getString(PKG,
"JsonOutput.Error.Writing"), e);
}
- // Close file
+ data.fileGenerator = null;
+ data.pendingFileItem = null;
+ data.fileItemCount = 0;
closeFile();
}
@@ -469,6 +528,8 @@ public class JsonEOutput extends
BaseTransform<JsonEOutputMeta, JsonEOutputData>
nc = HopJson.newMapper().getNodeFactory();
mapper = HopJson.newMapper();
+ // Items are written one by one: leave flushing to the buffers
+ fileMapper =
HopJson.newMapper().disable(SerializationFeature.FLUSH_AFTER_WRITE_VALUE);
first = false;
data.inputRowMeta = getInputRowMeta();
@@ -552,6 +613,19 @@ public class JsonEOutput extends
BaseTransform<JsonEOutputMeta, JsonEOutputData>
data.jsonKeyGroupItems = null;
}
+ // The file was not finished, for example after an error: flush what was
written and close any
+ // open array or object, so the partial file is at least well-formed.
+ if (data.fileGenerator != null) {
+ try {
+ data.fileGenerator.close();
+ } catch (IOException e) {
+ logError(BaseMessages.getString(PKG, "JsonOutput.Error.ClosingFile",
e.toString()));
+ setErrors(1);
+ }
+ data.fileGenerator = null;
+ data.pendingFileItem = null;
+ }
+
closeFile();
super.dispose();
}
diff --git
a/plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutputData.java
b/plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutputData.java
index 48404bf03c..f0c21d2335 100644
---
a/plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutputData.java
+++
b/plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutputData.java
@@ -17,6 +17,8 @@
package org.apache.hop.pipeline.transforms.jsonoutputenhanced;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.Writer;
import java.util.ArrayList;
@@ -39,7 +41,6 @@ public class JsonEOutputData extends BaseTransformData
implements ITransformData
public int[] fieldIndexes;
public int[] keysGroupIndexes;
public int nrRow;
- public List<ObjectNode> jsonItems;
public List<ObjectNode> jsonKeyGroupItems;
public String realBlocName;
@@ -51,6 +52,25 @@ public class JsonEOutputData extends BaseTransformData
implements ITransformData
public String openedFilename;
public boolean isWriteToFile;
+
+ /** The generated JSON goes to the output field (Output value, or both). */
+ public boolean isOutputValue;
+
+ /** Every row is its own item in the file and goes there straight away. */
+ public boolean streamFileRows;
+
+ /** Keep the items of the current group until the group is complete. */
+ public boolean collectGroupItems;
+
+ /** Writes the items of the file that is currently open. */
+ public JsonGenerator fileGenerator;
+
+ /** The first item of a file, held back until we know if the file needs an
array. */
+ public JsonNode pendingFileItem;
+
+ /** The number of items written to the file that is currently open. */
+ public int fileItemCount;
+
public String jsonSerialized;
public long jsonLength;
public Set<Integer> keyFields;
