question Re: Nifi Unzip files in Support Questions

Nifi Unzip files

lafi_oussama — Tue, 26 Sep 2023 09:55:19 GMT

Hello

I need help

I want to unzip file.zip with NiFi

I have used unpackContent processor but that didn't work because of my nifi version 1.13

I have also used executeStreamCommand to execute python script but I didn't found until now the solution

Can you help me

Thanks

Re: Nifi Unzip files

cotopaul — Tue, 26 Sep 2023 13:04:12 GMT

@lafi_oussama,

First of all, what is the compression format of the files you are trying to unzip?
What error did you receive when using UnpackContent?

Have you tried using CompressContent ( https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-standard-nar/1.12.1/org.apache.nifi.processors.standard.CompressContent/ )
You can configure the processed to either compress or decompress your files. You have multiple compression formats from which you can choose from.

Re: Nifi Unzip files

lafi_oussama — Tue, 26 Sep 2023 13:35:21 GMT

@cotopaul

The compression format is .zip.

The CompressContent doesn't work with zip format

Re: Nifi Unzip files

MattWho — Tue, 26 Sep 2023 14:19:27 GMT

@lafi_oussama

Have you tried upgrading to the latest version of Apache NiFi?
You have not shared the specific ERROR you are encountering with the UnpackContent processor.
You have not shared specific around what you tried using the ExecuteStreamCommand and Python scripts.
Any additional details you can provide about your zip files?

Thanks,
Matt

Re: Nifi Unzip files

lafi_oussama — Tue, 26 Sep 2023 14:31:52 GMT

I can't upgrade the version of NiFi.

the error I encountered with UnpackContent : Unable to unpack StandardFlowFileRecord[uuid=e94fb414-7fb3-441a-8294-89e2766633a4,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1695735090062-1030, container=default, section=6], offset=0, length=6002134],offset=0,name=outBI.zip,size=6002134] because it does not appear to have any entries; routing to failure

Example of python script that I have executed with ExecuteStreamCommand :

import zipfile
import sys
import gzip
import os
import io

zip_folder= sys.stdin.buffer.read()

with zipfile.ZipFile(io.BytesIO(zip_folder), 'r') as zip_folder:
for file_info in zip_folder.infolist():
file_name = file_info.filename
# Read the content of the file
file_content = zip_folder.read(file_name)
# Print the content to stdout (standard output)
sys.stdout.buffer.write(file_content)
sys.stdout.buffer.write(b'\n')
sys.stdout.flush()

Re: Nifi Unzip files

MattWho — Wed, 27 Sep 2023 13:29:43 GMT

@lafi_oussama

Does the zipfile actually contain files or only empty directories?

Re: Nifi Unzip files

joseomjr — Sun, 01 Oct 2023 20:29:19 GMT

If you can run this python code under Python without any external modules, you should be able to run it as a scripted processor and have everything happen inside of NiFi.

Re: Nifi Unzip files

lafi_oussama — Mon, 02 Oct 2023 09:11:31 GMT

@MattWho

yes the zipfile contains mutltiple floders that contain files

Re: Nifi Unzip files

lafi_oussama — Mon, 02 Oct 2023 09:11:10 GMT

@joseomjr

I didn't found the code that unzip the content of flowfile and send the unzipped files multiple flowfiles

Re: Nifi Unzip files

joseomjr — Sun, 08 Oct 2023 04:49:16 GMT

Created this Python ExecuteScript NiFi processor that extracts the files of a ZipFile (including those in subdirectories) into individual FlowFiles. It all happens inside of NiFi and not fully tested but it worked with a simple example in my lab.

"Script Body" below:

''' Extract Zip Files ''' from org.apache.commons.io import IOUtils from org.apache.nifi.processor.io import InputStreamCallback, OutputStreamCallback import zipfile from io import BytesIO class PyInputStreamCallback(InputStreamCallback): ''' InputStream Callback ''' def __init__(self): self.zip_file = None def process(self, input_stream): ''' Process our InputStream ''' zip_buffer = BytesIO(IOUtils.toByteArray(input_stream)) self.zip_file = zipfile.ZipFile(zip_buffer, "r") class PyOutputStreamCallback(OutputStreamCallback): ''' OutputStream Callback ''' def __init__(self, file): self.file = file def process(self, output_stream): ''' Process our OutputStream ''' output_stream.write(self.file.read()) flow_file = session.get() if flow_file: input_stream_callback = PyInputStreamCallback() output_stream_callback = PyOutputStreamCallback session.read(flow_file, input_stream_callback) zip_filename = flow_file.getAttribute("filename") zip_file = input_stream_callback.zip_file for name in (name for name in zip_file.namelist() if not name.endswith("/")): new_flow_file = session.create() new_flow_file = session.putAttribute(new_flow_file, "filename", name) new_flow_file = session.putAttribute(new_flow_file, "zip_filename", zip_filename) new_flow_file = session.write( new_flow_file, output_stream_callback(zip_file.open(name)) ) session.transfer(new_flow_file, REL_SUCCESS) zip_file.close() session.remove(flow_file)

Re: Nifi Unzip files

Shrink — Tue, 29 Apr 2025 05:40:32 GMT

@joseomjr
your code could not work at my end.
ChatGPT help me with below code and it is working at my end.

import java.util.zip.ZipInputStream import java.util.zip.ZipEntry import java.io.ByteArrayOutputStream def flowFile = session.get() if (!flowFile) return try { session.read(flowFile, { inputStream -> ZipInputStream zipIn = new ZipInputStream(inputStream) ZipEntry entry = zipIn.nextEntry while (entry != null) { if (!entry.isDirectory()) { ByteArrayOutputStream baos = new ByteArrayOutputStream() byte[] buffer = new byte[8192] int len while ((len = zipIn.read(buffer)) > 0) { baos.write(buffer, 0, len) } def contentBytes = baos.toByteArray() def newFlowFile = session.create(flowFile) newFlowFile = session.write(newFlowFile, { outStream -> outStream.write(contentBytes) } as OutputStreamCallback) newFlowFile = session.putAttribute(newFlowFile, 'filename', entry.getName()) session.transfer(newFlowFile, REL_SUCCESS) } zipIn.closeEntry() entry = zipIn.nextEntry } } as InputStreamCallback) session.remove(flowFile) } catch (Exception e) { log.error("Failed to unzip FlowFile due to: ${e.message}", e) session.transfer(flowFile, REL_FAILURE) }

I hope there is no security concern with this code!