Как манипулировать двумя потоками CSV-файлов в ExecuteScript с помощью Python? - PullRequest
Новая
       7

Как манипулировать двумя потоками CSV-файлов в ExecuteScript с помощью Python?

0 голосов
/ 27 июня 2018

В моем потоке я запрашиваю Hive, затем обновляю имена файлов, а затем хочу объединить эти csvs в одну книгу Excel с несколькими электронными таблицами. Мне удалось объединить два CSV-файла в одну книгу Excel с несколькими электронными таблицами, используя этот код. Как мне заставить скрипт использовать два файла из потока nifi вместо того, чтобы извлекать файлы из каталога на моем компьютере? Я видел, что это может сделать «flowFile = session.get ()», но эта строка захватывает оба потока файлов? 

import glob
import csv
import xlwt
import os
import xlsxwriter
import datetime
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import StreamCallback

wb = xlsxwriter.Workbook("combined_at%s.xlsx" %             datetime.datetime.now().strftime('%H-%M-%S'))

flowFile = session.get()

replacer = ",[]\"\"'\'"
worksheet = wb.add_worksheet("make")
worksheet2 = wb.add_worksheet("ownership")
worksheet3 = wb.add_worksheet("marital")
worksheet4 = wb.add_worksheet("drivers")
worksheet5 = wb.add_worksheet("vehicles")
worksheet6 = wb.add_worksheet("age")
worksheet7 = wb.add_worksheet("vyear")

def printHashedEmail(split_row, worksheet, index):
    for y in replacer:
        split_row[0] = split_row[0].replace(y, "")
    worksheet.write(index, 0, split_row[0])
    return;

def printOtherOnes(split_row, worksheet,index,non_changing_index):
    for y in replacer:
        split_row[non_changing_index] = split_row[non_changing_index].replace(y, "")
    worksheet.write(index, 1, split_row[non_changing_index])
    return;

with open("1.csv") as csv1:
    i = 0
    j = 0
    for row in csv1:
        split_row = row.split(",")
        if split_row[2] != "":
            printHashedEmail(split_row, worksheet, i)
            printOtherOnes(split_row,worksheet,i,2)
            i = i+1
        if split_row[3].strip() != "":
            printHashedEmail(split_row, worksheet2, j)
            printOtherOnes(split_row, worksheet2, j, 3)
            j = j+1


with open("2.csv") as csv1:
    i = 0; k = 0; j = 0; l = 0;m = 0;
    for row in csv1:
        split_row = row.split(",")
        if split_row[2] != "":
            printHashedEmail(split_row, worksheet3, i)
            printOtherOnes(split_row, worksheet3, i, 2)
            i = i + 1
        if split_row[3].strip() != "":
            printHashedEmail(split_row, worksheet4, j)
            printOtherOnes(split_row, worksheet4, j, 3)
            j = j + 1
        if split_row[5] != "":
            printHashedEmail(split_row, worksheet5, l)
            printOtherOnes(split_row, worksheet5, l, 5)
            l = l + 1
        if split_row[4].strip() != "":
            printHashedEmail(split_row, worksheet6, k)
            printOtherOnes(split_row, worksheet6, k, 4)
            k = k + 1
        if split_row[6].strip() != "":
            printHashedEmail(split_row,worksheet7,m)
            printOtherOnes(split_row, worksheet7, m, 6)
            m = m + 1

wb.close()

print("Done")

Вот этот поток Нифи

После манипуляции я бы хотел, чтобы файл excel вышел из ExecuteScriptProcessor, чтобы я мог делать с ним больше

1 Ответ

0 голосов
/ 27 июня 2018

проверить разные session.get() методы .

например session.get(2) попытается получить 2 первых файла из входящей очереди.

если у вас есть только один, вы можете позвонить session.rollback(), чтобы вернуть его в очередь.

но проблема здесь в том, что файлы в очереди могут быть в другом порядке, чем вы ожидали. просто представьте, что у вас есть 3 файла во входящей очереди.

с помощью session.get(FlowFileFilter filter) вы можете выбрать из входящей очереди 2 файлы, которые соответствуют некоторому атрибуту.

...