使用Nifi使用invokehttp将多个文件发布到API

问题描述

我有3个文件需要发布到API末尾.我正在使用FetchHDFS进程获取3个文件，我想将它们传递给API.我怎样才能通过他们.

I am having 3 files which i need to post to API end. I am fetching 3 files using FetchHDFS process and i want to pass them to API. How can i pass them.

输入:

3 files in HDFS
Content-Type: multipart/form-data

错误:

invokehttp.response.body
{ "message": "Multipart: Boundary not found (user: 'undefined')", "level": "error", "timestamp": "2019-12-11T09:59:05.464Z" }

尝试了流:

inputPort --> 3 FetchHDFS process to fetch 3 different file --> invokeHttps

curl commnad示例:

curl commnad sample:

curl -X POST "https://xxxxxx/xxxxx" -H "accept: application/json" -H "Content-Type: multipart/form-data" -F "[email protected];type=application/vnd.ms-excel" -F "[email protected];type=text/plain" -F "[email protected];type=application/vnd.ms-excel" -F "format=flat"

推荐答案

想法:从多个流文件构建多部分之前，需要将它们合并为一个.

The idea: Before building multipart from several flow files you need to merge them into one.

为此，将MergeContent处理器与Merge Format = TAR一起使用.

For this use the MergeContent processor with Merge Format = TAR.

然后使用ExecuteGroovyScript将TAR转换为multipart.

@Grab(group='org.apache.httpcomponents', module='httpmime', version='4.5.9')
@Grab(group='org.apache.commons', module='commons-compress', version='1.19')
import org.apache.commons.compress.archivers.tar.TarArchiveInputStream
import org.apache.http.entity.mime.MultipartEntityBuilder
import org.apache.http.entity.ContentType

def ff = session.get()
if(!ff)return

//delegate inputstream class to be able to set the `delegate` later
@groovy.transform.CompileStatic
class TarContentInputStream extends InputStream{
    @Delegate TarArchiveInputStream delegate
    @Override void close(){
        println "--------- try to close"
        if(!delegate.getNextTarEntry())delegate.close()
    }
}

def multipart = MultipartEntityBuilder.create()
def tarContent = new TarContentInputStream()

//iterate through TAR entries and build multipart
def tarInput=new TarArchiveInputStream(ff.read())
def tarEntry = tarInput.getNextTarEntry()
while (tarEntry != null) {
    //reference tarContent to be used as body
    multipart.addBinaryBody( tarEntry.getName(), tarContent, ContentType.DEFAULT_BINARY, tarEntry.getName() )
    tarEntry = tarInput.getNextTarEntry()
}
tarInput.close()
//write multipart content
ff.write{streamIn, streamOut->
    //set real input stream to be used as tar content
    tarContent.delegate = new TarArchiveInputStream(streamIn)
    assert tarContent.delegate.getNextTarEntry() //move to first entry
    multipart = multipart.build()
    multipart.writeTo(streamOut)
}

ff."mime.type" = multipart.getContentType().getValue()
ff.filename = ff.filename+".multipart"

REL_SUCCESS << ff

---

注意:

---

Note:

将3个测试文件合并为tar，上面的代码会产生类似以下内容:

for 3 test files combined into tar the code above produces something like:

--boundary
Content-Disposition: form-data; name="file1.txt"; filename="file1.txt"
Content-Type: application/octet-stream
Content-Transfer-Encoding: binary

file1 content
--boundary
Content-Disposition: form-data; name="file2.txt"; filename="file2.txt"
Content-Type: application/octet-stream
Content-Transfer-Encoding: binary

file2 content
--boundary
Content-Disposition: form-data; name="file3.txt"; filename="file3.txt"
Content-Type: application/octet-stream
Content-Transfer-Encoding: binary

file3 content
--boundary--

当前代码对输入流进行两次扫描:第一次-扫描tar条目，第二次-构建内容.

Currently code scans input stream two times: 1st - to scan tar entries, 2nd - to build content.

我认为可以重写代码以一次将tar转换为多部分...

I think it's possible to rewrite the code to convert tar to multipart in one shot...

这篇关于使用Nifi使用invokehttp将多个文件发布到API的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！