在apache NIFI中将xls文件转换为csv文件

Posted 2023-03-31

【中文标题】在apache NIFI中将xls文件转换为csv文件

【英文标题】：Convert xls files to csv files in apache NIFI

【发布时间】：2021-05-21 10:28:22

【问题描述】：

大家好，我正在尝试在 Apache NIFI 数据流中将 .xls 文件转换为 .csv。我尝试了很多解决方案都没有结果我什至尝试创建这样的脚本sccript

提前谢谢你

【问题讨论】：

您遇到了什么错误？

请阅读How to Ask 并使用tour。这个问题特别不清楚：“许多解决方案”，“没有结果” - 没有给我们暗示你做了什么，你投入了什么，你从中得到了什么，你期望什么。您可能想尝试提供minimal reproducible example。

【参考方案1】：

将您的ExecuteStreamCommand 配置为

并尝试在您的 python 脚本中使用以下代码，

import csv
import os
import sys
from io import StringIO, BytesIO
import pandas as pd
import xlrd
from pandas import ExcelFile

wb = xlrd.open_workbook(file_contents=sys.stdin.read(),logfile=open(os.devnull, 'w'))
excel_file_df = pd.read_excel(wb, sheet_name='Sheet1', index=False, index_col=0, encoding='utf-8',engine='xlrd')

#flowfile_content = ExcelFile(BytesIO(sys.stdin.read()))
#excel_file_df = pd.read_excel(flowfile_content, sheet_name='Sheet1', index=False, index_col=0, encoding='utf-8')

csv_data_rows = []
header_list = list(excel_file_df.columns.values)
temp_header_list = []

for field in header_list:
    temp = '"' + field +  '"'
    temp_header_list.append(temp)

header_row  = ','.join([str(elem) for elem in temp_header_list])
csv_data_rows.append(header_row)
is_header_row = True
for index, row in excel_file_df.iterrows():

    if is_header_row :
        is_header_row = False
        continue

    temp_data_list = []
    for item in row :
        #item = item.encode('utf-8', 'ignore').decode('utf-8')
        if hasattr(item, 'encode'):
            item = item.encode('ascii', 'ignore').decode('ascii')

        item = str(item)
        item = item.replace('\n', '')
        item = item.replace('",', '" ')
        if item == 'nan':
            item=''
        temp = '"' + str(item) + '"'
        temp_data_list.append(temp)

    data_row = ','.join([str(elem) for elem in temp_data_list])
    data_row = data_row
    csv_data_rows.append(data_row)

for item in csv_data_rows:
    sys.stdout.write("%s\r\n" % item)

【讨论】：

我试过你的解决方案，我得到了这个错误

TypeError: startswith first arg must be str or a tuple of str, not bytes

上面的代码中没有startsWith，你改了什么？在任何情况下，您都可以使用 decode("utf-8")

将字节更改为字符串

这正是代码中没有startsWith的问题，我不知道为什么

我认为问题出在 xlrd 库中

【参考方案2】：

你试过ConvertExcelToCSVProcessor吗？如果是这样并且它不起作用，您可以分享任何错误、日志等吗？

【讨论】：

是的，我试过了，但是这个处理器不只转换 xlsx 文件而不是 xls 我正在尝试使用 ExecuteStreamCommand 来运行 python 脚本，我收到一个错误，说 sys.stdin 是字节和不是str

stdin 是流文件

【参考方案3】：

    import csv
import os
import io
import sys
from io import StringIO, BytesIO
import pandas as pd
import xlrd
from pandas import ExcelFile
import petl as etl

#xls = etl.fromxls(sys.stdin)
#wb = xlrd.open_workbook(file_contents=xls,logfile=open(os.devnull, 'w', encoding='utf-8'))
excel_file_df = pd.read_excel(sys.stdin.buffer, sheet_name='Sheet1', index=False, index_col=0, encoding='utf-8',engine='xlrd')

#flowfile_content = ExcelFile(BytesIO(sys.stdin.read()))
#excel_file_df = pd.read_excel(flowfile_content, sheet_name='Sheet1', index=False, index_col=0, encoding='utf-8')

csv_data_rows = []
header_list = list(excel_file_df.columns.values)
temp_header_list = []

for field in header_list:
    temp = '"' + field +  '"'
    temp_header_list.append(temp)

header_row  = ','.join([str(elem) for elem in temp_header_list])
csv_data_rows.append(header_row)
is_header_row = True
for index, row in excel_file_df.iterrows():

    if is_header_row :
        is_header_row = False
        continue

    temp_data_list = []
    for item in row :
        #item = item.encode('utf-8', 'ignore').decode('utf-8')
        if hasattr(item, 'encode'):
            item = item.encode('ascii', 'ignore').decode('ascii')

        item = str(item)
        item = item.replace('\n', '')
        item = item.replace('",', '" ')
        if item == 'nan':
            item=''
        temp = '"' + str(item) + '"'
        temp_data_list.append(temp)

    data_row = ','.join([str(elem) for elem in temp_data_list])
    data_row = data_row
    csv_data_rows.append(data_row)

for item in csv_data_rows:
    sys.stdout.write("%s\r\n" % item)