Fabric을 활용한 통합 솔루션 구현 2

Fabric 개요

Column Storage의 장점과 사용 방법이다.

Column Storage의 대표적인 예시로 Parquet, ORC, Delta Lake, Iceberg가 있다.

 └ Parquet: 오픈소스 컬럼형 파일 포맷

 └ ORC: 하둡 생태계의 최적화된 컬럼형 포맷

 └ Delta Lake: Parquet 기반의 트랜잭션 지원 포맷

 └ Iceberg: 대규모 분석 데이터셋을 위한 테이블 포맷

Cardinarity가 높다는 뜻은 반복 처리가 좋다?

 

 

Lambda Architecture , Kappa Architecture는 둘 다 처리할 수 있는 구조

MS Fabric은 두 처리 방식을 모두 지원한다.

 

보안과 거버넌스 측면에서의 데이터다.

RBAC(Role Based Access Control): 역할 기반 접근 제어 권한 부여

ABAC(Attribute Based Access Control): 사용자나 리소스 속성 조건 기반 접근

MAC(Mandatory Access Control): 강제 접근 제어, 보안 정책에 따라 엄격한 계층 분리

DAC(Discretionary Access Control): 임의 접근 제어, 리소스 소유자가 권한을 직접 설정

 

Delta Table(델타 테이블)

Apache Spark와 Delta Lake에서 사용하는 고성능 테이블 포맷이다.

ACID 트랜잭션, 스키마 진화, 버전 관리, 타임트래블, 빠른 병합 및 업데이트를 지원하는 테이블이다.

보통 Databricks나 Azure Synapse, Fabric, EMR 등에서 사용한다.

DataFrame을 saveAsTable 메소드로 저장할 때, 2가지 형태의 delta 테이블로 저장할 수 있다.

 └ Managed Table(관리형 테이블): Fabric이 스키마 메타 데이터와 데이터 파일을 모두 관리한다.

 └ External Tanle(외부 테이블): 데이터를 외부 폴더(lakehouse)에 저장하고 메타데이터만 Fabric에서 관리한다.

 

분석계의 time over가 운영계로 영향을 끼쳐서는 안 된다. 이를 위해 Data Warehouse가 나왔다.

CREATE TABLE AS SELECT(CTAS)나 SELECT INTO 같은 명령으로 대량의 데이터를 효율적으로 처리한다.

 

	전통적 DW	Hadoop / Spark	MS Fabric
아키텍쳐	On premise / IaaS	분산 컴퓨팅	Serverless, Cloud Native
확장성	제한, 수직	높음, 수평	무제한, 자동 확장
데이터 처리	정형 데이터 중심	비정형 데이터 특화	모든 데이터 지원
관리 부담	높음	매우 높음	낮음
ETL 요구사항	복잡한 ETL	ELT	Delta 기반 통합 처리
비용 모델	높은 고정 비용	높은 인프라 비용	사용량 기반 가변

Data warehouse 비교 표다. 실제로 이런 지는 확인해봐야 한다.

 

Fabric Dataflow

Dataflow 실습용 lakehouse를 생성한다. 그리고 Get data에서 Dataflow를 생성한다.

 

Dataflow가 생기면 csv 파일을 업로드한다.

 

https://raw.githubusercontent.com/MicrosoftLearning/dp-data/main/orders.csv 링크에서 다운받는다.

 

몇 월인지 구분할 수 있는 새로운 column을 하나 만든다.

 

이후 이어줄 lakehouse를 선택한다.

만약 이미 Data destination이 있다면 삭제 후 연결하는 것이 가능하다.

 

자신이 연결할 lakehouse와 선택하고 테이블 이름을 설정한다.

 

Dataflow의 흐름은 View에서 확인할 수 있다.

 

Pipeline을 만들고 방금 만든 dataflow를 넣어서 실행할 수 있다.

Dataflow를 직접 실행하나, pipeline에 넣어서 실행하나 기능적으로 동일하다.

단지 이런 방법의 흐름도 있음을 보여주고, 추가 작업이 있는 경우에 사용 가능하다는 것을 시사한다.

 

파이프라인이든 흐름이든 실행하면 테이블이 들어온다.

 

Fabric Delta Table

https://github.com/MicrosoftLearning/dp-data/raw/main/products.cs 링크에서 csv를 받아 올린다.

 

각각의 코드를 사용하여 managed table과 external table로 하나의 df를 저장하였다.

 

from pyspark.sql.types import StructType, IntegerType, StringType, DoubleType


# define the schema
schema = (
    StructType()
    .add("ProductID", IntegerType(), True)
    .add("ProductName", StringType(), True)
    .add("Category", StringType(), True)
    .add("ListPrice", DoubleType(), True)
)

df = spark.read.format("csv").option("header","true").load("abfss://c0acebfe-272a-4eec-8c1b-41d9a825c52e@onelake.dfs.fabric.microsoft.com/cb839aa2-6567-4fde-b6a7-523612d54be1/Files/products/products.csv")
# display(df)

# Make Managed Table
df.write.format("delta").saveAsTable("managed_products")

# Make External Table
files_folder_path = "abfss://c0acebfe-272a-4eec-8c1b-41d9a825c52e@onelake.dfs.fabric.microsoft.com/cb839aa2-6567-4fde-b6a7-523612d54be1/Files/"
df.write.format("delta").saveAsTable("external_products", path=files_folder_path + "external_products")

사용한 코드는 위와 같다.

이때 external table로 저장할 때는 "your_lakehouse_id/Files/your_folder_name"라고 명시해주거나

Files를 우클릭하여 Copy ABFS path로 경로를 가져와서 사용하면 된다.

 

Managed table과 external table 모두 정상적으로 조회가 된다.

이때 data_type을 늘려서 확인해보면 각 저장소가 상이함을 볼 수 있다.

 

둘 다 DROP으로 삭제했을 때, external table은 Filers 폴더에서 사라지지 않음을 볼 수 있다.
즉 External table의 메타데이터는 지워졌어도 데이터 파일은 남아있다.

 

트랜잭션을 확인할 겸 다시 delta table을 만들어준다.

이번에는 SQL을 사용하여 만들고, Files 폴더에 있는 external_products 데이터를 사용하여 만든다.

USING은 delta table로 생성하는 명령, LOCATION은 해당 경로의 데이터로 테이블을 생성하는 명령이다.

 

확인을 위해 간단한 UPDATE 구문을 실행했다. 이때 트랜잭션을 확인하기 위함이니 어떤 SQL 명령이든 상관없다.

이후 DESCRIBE HISTORY your_table_name;으로 트랜잭션 로그를 볼 수 있다.

여기서는 version 0에 테이블 생성, version 1에 테이블 업데이트 작업이 들어있다.

 

delta_table_path = 'Files/external_products'
# Get the current data
current_data = spark.read.format("delta").load(delta_table_path)
display(current_data)

# Get the version 0 data
original_data = spark.read.format("delta").option("versionAsOf", 0).load(delta_table_path)
display(original_data)

위 코드를 통해 버전별로 table을 확인할 수 있다.

 

Fabric Streaming Data

Delta Lake는 스트리밍 데이터를 지원한다.

Delta table은 Spark 구조적 스트리밍 API로 생성한 데이터 스트림 sink나 source가 될 수 있다.

위 코드처럼 스트리밍 소스를 설정한다.

 

from notebookutils import mssparkutils
from pyspark.sql.types import *
from pyspark.sql.functions import *

# Create a folder
inputPath = 'Files/data/'
mssparkutils.fs.mkdirs(inputPath)

# Create a stream that reads data from the folder, using a JSON schema
jsonSchema = StructType([
    StructField("device", StringType(), False),
    StructField("status", StringType(), False)
])
iotstream = spark.readStream.schema(jsonSchema).option("maxFilesPerTrigger", 1).json(inputPath)

# Write some event data to the folder
device_data = '''{"device":"Dev1","status":"ok"}
{"device":"Dev1","status":"ok"}
{"device":"Dev1","status":"ok"}
{"device":"Dev2","status":"error"}
{"device":"Dev1","status":"ok"}
{"device":"Dev1","status":"error"}
{"device":"Dev2","status":"ok"}
{"device":"Dev2","status":"error"}
{"device":"Dev1","status":"ok"}'''

mssparkutils.fs.put(inputPath + "data.txt", device_data, True)
print("Source stream created...")

사용한 코드는 위와 같다.

 

스트리밍 source를 지정한 다음, 위 화면처럼 스트림 데이터를 delta table로 보낸다.

Delta table을 query를 사용하여 IotDeviceData 테이블로 받는다.

 

# Write the stream to a delta table
delta_stream_table_path = 'Tables/iotdevicedata'
checkpointpath = 'Files/delta/checkpoint'
deltastream = iotstream.writeStream.format("delta").option("checkpointLocation", checkpointpath).start(delta_stream_table_path)
print("Streaming to delta sink...")

사용한 코드는 위와 같고, 코드에 대한 설명은 아래와 같다.

iotstream.writeStream: 읽기 스트림(iotstream)을 쓰기 스트림으로 전환한다.

.format("delta"): 스트림의 출력 형식을 Delta로 지정한다.

.option("checkpointLocation", ...): 스트림 처리 상태를 저장할 체크포인트 위치를 지정한다.

 └ 이는 스트림 장애 발생 시 중단된 지점부터 안정적으로 재시작하기 위해 필수다.

.start(delta_stream_table_path): 스트림 쓰기를 Tables/iotdevicedata 경로로 시작한다.

 └ Tables IotDeviceData 경로에 쓰기 때문에 IotDeviceData라는 이름의 테이블이 자동으로 생성한다.

 

데이터를 추가로 더 받아오는 것도 가능하다.

 

# Add more data to the source stream
more_data = '''{"device":"Dev1","status":"ok"}
{"device":"Dev1","status":"ok"}
...'''
mssparkutils.fs.put(inputPath + "more-data.txt", more_data, True)

데이터를 더 받아오는 코드는 위와 같다.

 

Fabric Data Warehouse

Warehouse를 생성한다.