Azure Databricks, 데이터 수집하기(파일 업로드, 자동 로더, DBFS..)

1. 로컬 데이터 파일 업로드하여 테이블 만들기

사이즈가 작은 CSV or TSV 파일을 UI통해서 직접 업로드하여 테이블을 생성할 수 있다.

- 파일 업로드 방법

  - New > File Upload

  - New > Add data > upload Data

  - Data > (tab) Database Tables > Add data > Upload data

 

2. Delta Lake로 데이터를 수집하도록 자동 로더 구성하기

자동 로더란? - Azure Databricks | Microsoft Learn

자동 로더란? - Azure Databricks

 

 

https://youtu.be/Wte44wRZKDk

Databricks는 증분 데이터 수집을 위해 자동 로더를 사용할 것을 권장. 자동 로더는 클라우드 개체 스토리지에 도착하는 새 파일을 자동으로 감지하고 처리한다.

Databricks에서는 Delta lake를 사용하여 데이터를 저장하는 것이 좋다.

Delta Lake는 ACID 트랜잭션을 제공하고 데이터 레이크하우스를 가능하게 하는 오픈 소스 스토리지 레이어.

Delta Lake는 Databricks에서 만든 테이블의 기본 형식.

# Import functions
from pyspark.sql.functions import input_file_name, current_timestamp

# Define variables used in code below
file_path = "/databricks-datasets/structured-streaming/events"
username = spark.sql("SELECT regexp_replace(current_user(), '[^a-zA-Z0-9]', '_')").first()[0]
table_name = f"{username}_etl_quickstart"
checkpoint_path = f"/tmp/{username}/_checkpoint/etl_quickstart"

# Clear out data from previous demo execution
spark.sql(f"DROP TABLE IF EXISTS {table_name}")
dbutils.fs.rm(checkpoint_path, True)

# Configure Auto Loader to ingest JSON data to a Delta table
(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load(file_path)
  .select("*", input_file_name().alias("source_file"), current_timestamp().alias("processing_time"))
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable(table_name))

 

3. 스토리지 계정으로 비구조적 데이터 수집하기

Notebook 생성하고 Cluster 연결 : +New > Notebook

Spark 통해서 컨테이너 탑재

application id : Azure Active Directory > 앱 등록 > 개요 > 애플리케이션(클라이언트)ID

secret : Azure Active Directory > 앱 등록 > 인증서 및 암호 > 클라이언트 암호의 [값]

tenant id: Azure Active Directory > 앱 등록 > 개요 > 디렉터리(테넌트)ID

⚠경로, 파일 위치 확인해서 오류나지 않도록 주의한다.

configs = {"fs.azure.account.auth.type": "OAuth",
       "fs.azure.account.oauth.provider.type": "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",
       "fs.azure.account.oauth2.client.id": "<application-id>", --
       "fs.azure.account.oauth2.client.secret": "<client-secret>",
       "fs.azure.account.oauth2.client.endpoint": "https://login.microsoftonline.com/<tenant-id>/oauth2/token",
       "fs.azure.createRemoteFileSystemDuringInitialization": "true"}

dbutils.fs.mount(
source = "abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/",
mount_point = "/mnt/flightdata",
extra_configs = configs)

⚠ 스토리지 계정의 액세스 키를 통해서 바로 스토리지에 붙는 방법도 있으나, 보안에 취약하므로 권장하지 않는다.

spark.conf.set("fs.azure.account.key.<storage-account-name>.dfs.core.windows.net","<storage-account-access-key>")
sdf = spark.read.csv(path='abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/nyc-taxi.csv',header=True)
display(sdf)

 

4. DBFS(Databricks File System)

 

DBFS : Databricks 작업 영역에 탑재되고 Databricks 클러스터에서 사용할 수 있는 분산 파일 시스템

 

DBFS(Databricks 파일 시스템)란? - Azure Databricks | Microsoft Learn

DBFS(Databricks 파일 시스템)란? - Azure Databricks

DBFS에 파일 업로드

New > Data > DBFS > Upload File

 

⚠️ DBFS Target Directory 의 optional 부분에 추가하고 싶은 새폴더명을 입력 후 select를 누르면 확인창이 열리고 Save 가능하다. 입력하지 않으면, DBFS/FileStore/tables 하위에 파일이 업로드된다.

 

⚠️ DBFS루트에는 작업 영역의 모든 사용자가 접근할 수 있기 때문에, 중요한 정보를 이곳에 저장하지 않도록 한다.

Hive metastore에 있는 관리되는 테이블의 기본위치는 DBFS루트 >> 관리되는 테이블을 만드는 최종 사용자가 DBFS루트에 쓰기를 수행하지 않도록 하려면, Hive metastore에 데이터베이스를 만들 때 외부 스토리지 위치를 선언한다. (Databricks에서는 관리되는 테이블에 대해 Unity Catalog를 사용하는 것이 좋다(기본적으로 보안 스토리지 위치가 사용됨))

DBFS에 업로드된 파일 선택하여 테이블 생성하기

New > Data > DBFS > DBFS

⚠️ 테이블 생성은 UI 또는 Notebook을 통해서 가능하다. UI를 사용하면 외부테이블만 만들 수 있다.

 

Databricks