Python 라이브러리를 활용한 Azure Blob 파일 획득 예제

Python 라이브러리를 활용해서 Azure Blob 스토리지의 Blob 파일을 획득하는 예제를 진행해보겠다. 이를 위해 azure-storage-blob 이라는 이름의 패키지를 설치한다. 

pip install azure-storage-blob

사용할 때에는 azure.storage.blob으로 임포트한다. 단, 패키지 사이즈가 크므로 사용할 라이브러리만 이름으로 임포트하여 사용하는 것을 추천한다.

from azure.storage.blob import *

 

1. MS Azure Blob File 획득

1-1. MS Azure Blob File 이름 리스트 획득

컨테이너 하위의 blob 파일 리스트에 대한 정보는 Python azure-storage-blob의 BlobServiceClient 클래스가 지원한다. 다만, 아래 코드에서 list_blobs 메소드를 통해 얻은 객체 'blobs_list'는 ItemPaged 타입의 객체이다. 해당 객체의 name 속성을 통해 파일의 이름을 얻을 수 있다.

 

// code placeholder
// # Instantiate a BlobServiceClient using a connection string
from azure.storage.blob import BlobServiceClient

// # Import Configuration Variables
// # If there is a problem to configure access info of MS Azure,
// # then check the configuration file ./utils/iddx_configs.py
import utils.iddx_configs as IddxConfigs
AZURE_CCKR_STORAGE_CONNECTION_STRING = 'DefaultEndpointsProtocol={0};AccountName={1};AccountKey={2};EndpointSuffix={3}'.format(IddxConfigs.IDDX_SOLMA_CCKR_DEFAULTENDPOINTSPROTOCOL, IddxConfigs.IDDX_SOLMA_CCKR_ACCOUNTNAME, IddxConfigs.IDDX_SOLMA_CCKR_ACCOUNTKEY, IddxConfigs.IDDX_SOLMA_CCKR_ENDPOINTSUFFIX)
CONTAINER_NAME = 'my_contatiner'
FILENAME_STARTWITH = 'my_blob_file_name_start_with_this_text'

// # Instantiate a BlobServiceClient
blob_service_client = BlobServiceClient.from_connection_string(AZURE_CCKR_STORAGE_CONNECTION_STRING)

container_client = blob_service_client.get_container_client(CONTAINER_NAME)
blobs_list = container_client.list_blobs() // # ItemPaged 타입 객체
blob_name_list = [_.name for _ in blobs_list] // # 해당 container 하위 모든 blob files 이름 리스트

blobs_list = container_client.list_blobs(FILENAME_STARTWITH) // # ItemPaged 타입 객체
blob_name_list = [_.name for _ in blobs_list] // # 해당 container 하위 모든 blob files 중 'FILENAME_STARTWITH'로 파일명이 시작하는 이름 리스트

 

Blob에 대한 Multi-Primary Key를 name, container, creation_time 를 활용해야 한다. (blob을 그대로 document로 사용할 경우)

BlobServiceClient의 list_blobs 메소드를 활용하여 생성한 Item Paged 타입의 객체는 아래와 같은 속성과 값을 갖는다. Azure Blob 데이터를 저장, 활용하려는 경우, Blob의 인덱스 태그를 사용하여 관리하는 것이 효율적이다. 특정한 형태의 제약이 없으므로, 인덱스 태그를 업데이트하여 개체를 동적으로 분류하는 과정이다. 

 

인덱스 태그를 활용할 수 없을 때에는 Multi-Primary Key를 활용하자.

하지만 클라우드 제품에 따라 설계 시 인덱스 태그를 제공하지 않을 수도 있다(인덱스 태그는 현재 구독 버전에서 무료 사용임에도 본인이 근무하는 회사의 클라우드 제품 개발 시 인덱스 태그를 따로 지정하지 않은 듯 하다...). 따라서 Blob 데이터를 가져올 때, Blob 데이터를 이미 수집했는지 확인하기 위해서는 Multi-Primary Key를 활용해야 하는 상황이다. 파일의 생성 여부, 업데이트 여부를 확인할 수 있는 최소한의 Key 선택이 중요하다. 다만 etag, last_modified를 선택하면 blob에 접근할때마다 이 값이 갱신되므로 데이터 무결성을 방해한다. 인덱스로 파일 리스트 업에 드는 비용을 최소화할 수 있으나, 이가 없는 경우 속성을 활용하여 해당 Blob을 로드할 것인지 여부를 결정할 수 있다. 데이터를 제공하는 솔루션에 대한 정보는 index_name으로 제공하므로 제외하고 container, country에 대한 정보도 Key로 들어가야 한다. 

 

// code placeholder
    Properties for Each ItemPaged Class Object
    from
        container_client = blob_service_client.get_container_client()
        blobs_list = container_client.list_blobs()
    {
        'name': 'my_blob_file_name.json',
        'container': 'status', 
        'snapshot': None, 
        'version_id': None, 
        'is_current_version': None, 
        'blob_type': <BlobType.BlockBlob: 'BlockBlob'>, 
        'metadata': {}, 
        'encrypted_metadata': None, 
        'last_modified': datetime.datetime(2020, 1, 01, 1, 0, tzinfo=datetime.timezone.utc), 
        'etag': # something something, 
        'size': 1632605, 
        'content_range': None, 
        'append_blob_committed_block_count': None, 
        'is_append_blob_sealed': None, 
        'page_blob_sequence_number': None, 
        'server_encrypted': True, 
        'copy': {'id': None, 'source': None, 'status': None, 'progress': None, 'completion_time': None, 'status_description': None, 'incremental_copy': None, 'destination_snapshot': None}, 
        'content_settings': {'content_type': 'application/octet-stream', 'content_encoding': None, 'content_language': None, 'content_md5': None, 'content_disposition': None, 'cache_control': None}, 
        'lease': {'status': 'unlocked', 'state': 'available', 'duration': None}, 
        'blob_tier': None, 
        'rehydrate_priority': None, 
        'blob_tier_change_time': None, 
        'blob_tier_inferred': None, 
        'deleted': None, 
        'deleted_time': None, 
        'remaining_retention_days': None, 
        'creation_time': datetime.datetime(2020, 1, 01, 0, 56, 53, tzinfo=datetime.timezone.utc), 
        'archive_status': None, 
        'encryption_key_sha256': None, 
        'encryption_scope': None, 
        'request_server_encrypted': None, 
        'object_replication_source_properties': [], 
        'object_replication_destination_policy': None, 
        'last_accessed_on': None, 
        'tag_count': None, 
        'tags': None, 
        'immutability_policy': {'expiry_time': None, 'policy_mode': None}, 
        'has_legal_hold': None, 
        'has_versions_only': None  
    }

 

1-2. MS Azure Blob File 이름을 통해 개별 Blob File의 정보를 획득

 컨테이너 하위의 개별 blob 파일에 대한 정보는 Python azure-storage-blob의 BlobClient 클래스가 지원한다. 

 

// code placeholder
// # Instantiate a BlobClient using a connection string
from azure.storage.blob import BlobClient

// # Import Configuration Variables
// # If there is a problem to configure access info of MS Azure,
// # then check the configuration file ./utils/iddx_configs.py
import utils.iddx_configs as IddxConfigs
AZURE_CCKR_STORAGE_CONNECTION_STRING = 'DefaultEndpointsProtocol={0};AccountName={1};AccountKey={2};EndpointSuffix={3}'.format(IddxConfigs.IDDX_SOLMA_CCKR_DEFAULTENDPOINTSPROTOCOL, IddxConfigs.IDDX_SOLMA_CCKR_ACCOUNTNAME, IddxConfigs.IDDX_SOLMA_CCKR_ACCOUNTKEY, IddxConfigs.IDDX_SOLMA_CCKR_ENDPOINTSUFFIX)
CONTAINER_NAME = 'my_container_name'

// # Instantiate a BlobClient and Get Each File Info
blob_file_name = blob_name_list[0]
blob_client = BlobClient.from_connection_string(AZURE_CCKR_STORAGE_CONNECTION_STRING, container_name=CONTAINER_NAME, blob_name=blob_file_name)

snapshot_blob = blob_client.create_snapshot()
// # snapshot_blob = {'snapshot': '2022-01-01T08:18:30.1243451Z', 'etag': '~~~', 'last_modified': datetime.datetime(2022, 1, 1, 8, 18, 29, tzinfo=datetime.timezone.utc), 'client_request_id': '~~~~~', 'request_id': '~~~~~', 'version': '2020-10-02', 'version_id': None, 'date': datetime.datetime(2022, 1, 6, 8, 18, 29, tzinfo=datetime.timezone.utc), 'request_server_encrypted': False}

 

 

---

2. 개별 Blob File의 데이터 확인

2-1. File 데이터를 확인하는 방법

Blob File에는 어떤 데이터가 들어있는지 확인하기 위해 일단 로컬에 개별 파일 하나를 다운로드 받겠다.

// code placeholder
local_filepath_name = f"{current_file_path}/blobdownload/{blob_file_name.split('/')[-1]}"
with open(local_filepath_name, 'wb') as local_blob_file:
     download_stream = blob_client.download_blob()
     local_blob_file.write(download_stream.readall())

 

 

2-2. 데이터 저장소의 결정

Blob은 비정형 데이터이므로 SQL DB에 저장하는 시나리오는 제외하고 저장소를 결정하겠다. 현재 서버에서 사용 중인 저장소는 파일시스템과 엘라스틱서치가 있다.

저장소	서버 파일 시스템	서버 엘라스틱 서치
장점	빠르게 데이터 획득 가능	kibana로 데이터 탐색이 쉽게 가능
단점	데이터 가시화가 되지 않음	관계형 DB에 비해 스터디 시간이 필요함

 

현재 Blob File은 개별 json 객체가 2차원 이상이다. 이런 형태의 데이터는, 엘라스틱 서치에 매핑 처리하여 저장해야 하지만, 현재 Task 정책 및 Kibana 등의 처리를 고려할 때, 2차원 객체로 전처리하여 저장하도록 한다. 객체를 전처리 하기 전에 서버 파일 시스템에 blob 파일 자체를 iddx_cloudanalytics 하위 폴더에 저장한다. (paramico 라이브러리 이용하여 원격 서버에 접속하여 직접 파일 업로드할 수 있을 것 같음)

 

 

추가적인 첨언

Python 라이브러리를 활용하여 Azure Blob 파일을 획득하는 경우 그 속도가 매우 느리기 때문에 분석 파이프라인을 구축할 때 지속적인 고민이 필요하다. 파일을 획득하여 바로 모델링에 활용할 것인지, 아니면 전처리를 하여 저장한 뒤에 다시 모델링에 활용할 것인지 등.. azure-blob-storage 패키지로 한 번에 가져올 수 있는 데이터의 양, 시간을 확인하고, 어떤 주기로 azure 에 접근하여 데이터를 획득할 것인지, 시스템 부하 혹은 클라우드 사용량, 사용 비용에 무리가 되지는 않는지를 테스트를 통해 검증하는 것이 우선이다. 분석 부서에서 이를 활용하는 것에 대해 개발/비용 부서와 사전 협의는 분명히 필요하며, 용량, 비용 과다 부과가 우려되는 경우 데이터 파이프라인 설계 시 재협의를 진행하는 것이 필수적이다. 아마도 가장 좋은 방법은 개발 부서에서 데이터를 쌓는 시점에서 분석이 가능하도록 파이프라인을 구축(FastAPI 등으로 분석 방법을 제공)하는 것인데, 부서 간 업무 KPI가 다르므로 회사마다 이렇게 진행하는 것은 조금 어려울 수 있는 것이 현실임을 감안하여, 가장 합리적인 선택을 하는 것이 중요하다. 중간에 위치해 있기 때문에, 언제나 을이지만 데이터에 접근하게 해주는 것만으로도 행복한 것이 데이터 싸이언티스트의 현실이랄까....ㅎㅎㅎ

 

#. 레퍼런스

Python Library: azure-storage-blob / ContainerClient Class

Python Library: azure-storage-blob / BlobClient Class

Python Library: azure-storage-blob / BlobServiceClient Class