Создайте блоб из блоков с Azure in Python - PullRequest
Новая
       69

Создайте блоб из блоков с Azure in Python

0 голосов
/ 03 апреля 2020

Попытка загрузить большой двоичный объект как df, сохранить его в памяти, а затем добавить к нему флаг столбца. Затем загрузите новый файл с тем же именем. Df_str - это загруженный df в памяти в виде строки. Размер df превышает 64 МБ, поэтому create_blob_from_ * не работает. Загрузка и изменение df работает нормально, но загрузка выдает ошибку. "Нет идентификатора в списке". 

import logging
import azure.functions as func
import io
import datetime
import pandas as pd
import base64
import uuid
from azure.storage import *
import base64
import os
from azure.storage.blob import BlockBlobService

def increment(value: str)-> str:
    value_as_int = int(value)
    value_as_int += 1
    new_value = ""
    for i in range(3):
        new_value += "0"
    new_value += str(value_as_int)
    new_value = new_value[-4:]

    return new_value


def main(req: func.HttpRequest) -> func.HttpResponse:
    try:
        logging.info('Python HTTP trigger function processed a request.')

        STORAGEACCOUNTNAME = "XXXX"
        STORAGEACCOUNTKEY = "YYYYY"

        acc_name = STORAGEACCOUNTNAME
        acc_key = STORAGEACCOUNTKEY
        container = "CONTAINER_NAME"
        blob_name = "FILENAME.txt"

        blob = BlockBlobService(acc_name, acc_key)

        # Change df stored in memory
        df = pd.read_csv(io.StringIO(df_str), sep="|")
        df = df.assign(flag=False)

        input.close()

        block_ids = []

        write_to_chunk_size = 500
        new_index = "0001"
        with io.BytesIO(df.to_csv(sep="|").encode()) as f:
            while True:
                write_data = f.read(write_to_chunk_size)
                if write_data:
                    length = len(write_data)
                    block_id = base64.urlsafe_b64encode((new_index).encode())
                    blob.put_block(container, blob_name, write_data, block_id)
                    block_ids.append(test(block_id))
                    new_index = increment(new_index)
                else:
                    break

        blob.put_block_list(container_name=container, blob_name=blob_name, block_list=block_ids)

        return func.HttpResponse("DONE")
    except Exception as error:
        func.HttpResponse(error, status_code=500)
...