Pyspark 사용예 (HDFS)

 

from pyspark.sql import SparkSession

# Spark 세션 생성
spark = SparkSession.builder \
    .appName("ETL Example") \
    .getOrCreate()

# 1. 데이터 추출 (Extract)
# HDFS에서 CSV 파일 읽기
input_path = "hdfs://namenode:8020/path/to/input/data.csv"
df = spark.read.csv(input_path, header=True, inferSchema=True)

# 2. 데이터 변환 (Transform)
# 예: 특정 열의 값을 변경하거나 새로운 열 추가
transformed_df = df.withColumn("new_column", df["existing_column"] * 2) \
                   .filter(df["filter_column"] > 100)

# 3. 데이터 적재 (Load)
# 변환된 데이터를 HDFS에 저장 (Parquet 형식으로 저장)
output_path = "hdfs://namenode:8020/path/to/output/transformed_data.parquet"
transformed_df.write.parquet(output_path, mode="overwrite")

# Spark 세션 종료
spark.stop()