Data Lake

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303

import boto3
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, isnan, count
import logging
from datetime import datetime

class DataLakeManager:
    """
    데이터 레이크 관리를 위한 클래스
    주요 기능: 데이터 수집, 저장, 품질 검증, 카탈로그 관리
    """
    
    def __init__(self, bucket_name, region='us-east-1'):
        """
        데이터 레이크 매니저 초기화
        Args:
            bucket_name: S3 버킷 이름
            region: AWS 리전
        """
        self.bucket_name = bucket_name
        self.s3_client = boto3.client('s3', region_name=region)
        self.glue_client = boto3.client('glue', region_name=region)
        
        # Spark 세션 초기화 - 데이터 처리 엔진
        self.spark = SparkSession.builder \
            .appName("DataLakeProcessor") \
            .config("spark.sql.adaptive.enabled", "true") \
            .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
            .getOrCreate()
        
        # 로깅 설정 - 운영 모니터링을 위한 로그
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
    
    def ingest_batch_data(self, source_path, destination_key, data_format='parquet'):
        """
        배치 데이터 수집 및 저장
        Args:
            source_path: 소스 데이터 경로
            destination_key: S3 저장 경로
            data_format: 데이터 형식 (parquet, json, csv)
        """
        try:
            # 데이터 읽기 - 다양한 형식 지원
            if data_format == 'parquet':
                df = self.spark.read.parquet(source_path)
            elif data_format == 'json':
                df = self.spark.read.json(source_path)
            else:  # CSV
                df = self.spark.read.option("header", "true").csv(source_path)
            
            # 데이터 품질 검증 - 결측치 및 중복 확인
            quality_metrics = self._validate_data_quality(df)
            self.logger.info(f"Data quality metrics: {quality_metrics}")
            
            # 메타데이터 추가 - 데이터 계보 추적을 위한 정보
            df_with_metadata = df.withColumn("ingestion_timestamp", 
                                           lit(datetime.now().isoformat()))
            
            # 파티션 기반 저장 - 성능 최적화를 위한 파티셔닝
            partition_column = self._determine_partition_column(df_with_metadata)
            
            # S3에 저장 - Raw 계층에 원시 데이터 저장
            s3_path = f"s3a://{self.bucket_name}/raw/{destination_key}"
            
            if partition_column:
                df_with_metadata.write.mode("append") \
                    .partitionBy(partition_column) \
                    .parquet(s3_path)
            else:
                df_with_metadata.write.mode("append").parquet(s3_path)
            
            # 메타데이터 카탈로그 업데이트
            self._update_glue_catalog(destination_key, df_with_metadata.schema)
            
            self.logger.info(f"Successfully ingested {df.count()} records to {s3_path}")
            
        except Exception as e:
            self.logger.error(f"Error ingesting data: {str(e)}")
            raise
    
    def _validate_data_quality(self, df):
        """
        데이터 품질 검증 함수
        Returns:
            dict: 품질 메트릭 정보
        """
        total_rows = df.count()
        
        # 컬럼별 결측치 계산
        null_counts = df.select([count(when(col(c).isNull() | isnan(col(c)), c)).alias(c) 
                               for c in df.columns]).collect()[0].asDict()
        
        # 중복 레코드 수 계산
        distinct_rows = df.distinct().count()
        duplicate_rows = total_rows - distinct_rows
        
        return {
            'total_rows': total_rows,
            'duplicate_rows': duplicate_rows,
            'null_counts': null_counts,
            'duplicate_percentage': (duplicate_rows / total_rows) * 100 if total_rows > 0 else 0
        }
    
    def _determine_partition_column(self, df):
        """
        최적의 파티션 컬럼 결정
        Returns:
            str: 파티션 컬럼명 또는 None
        """
        # 날짜/시간 컬럼 우선 선택
        date_columns = [col for col in df.columns 
                       if 'date' in col.lower() or 'time' in col.lower()]
        
        if date_columns:
            return date_columns[0]
        
        # 카디널리티가 적절한 범위의 컬럼 선택 (100-10000)
        for col_name in df.columns:
            if df.select(col_name).distinct().count() in range(100, 10000):
                return col_name
        
        return None
    
    def _update_glue_catalog(self, table_name, schema):
        """
        AWS Glue 카탈로그 업데이트 - 메타데이터 관리
        Args:
            table_name: 테이블명
            schema: 스키마 정보
        """
        try:
            # Spark 스키마를 Glue 스키마로 변환
            glue_columns = []
            for field in schema.fields:
                glue_columns.append({
                    'Name': field.name,
                    'Type': self._spark_to_glue_type(field.dataType)
                })
            
            # Glue 테이블 정의
            table_input = {
                'Name': table_name.replace('/', '_'),
                'StorageDescriptor': {
                    'Columns': glue_columns,
                    'Location': f"s3://{self.bucket_name}/raw/{table_name}/",
                    'InputFormat': 'org.apache.hadoop.mapred.TextInputFormat',
                    'OutputFormat': 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat',
                    'SerdeInfo': {
                        'SerializationLibrary': 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
                    }
                }
            }
            
            # 테이블 생성 또는 업데이트
            try:
                self.glue_client.create_table(
                    DatabaseName='data_lake_catalog',
                    TableInput=table_input
                )
            except self.glue_client.exceptions.AlreadyExistsException:
                self.glue_client.update_table(
                    DatabaseName='data_lake_catalog',
                    TableInput=table_input
                )
            
            self.logger.info(f"Updated Glue catalog for table: {table_name}")
            
        except Exception as e:
            self.logger.warning(f"Failed to update Glue catalog: {str(e)}")
    
    def _spark_to_glue_type(self, spark_type):
        """
        Spark 데이터 타입을 Glue 데이터 타입으로 변환
        """
        type_mapping = {
            'StringType': 'string',
            'IntegerType': 'int',
            'LongType': 'bigint',
            'DoubleType': 'double',
            'BooleanType': 'boolean',
            'TimestampType': 'timestamp'
        }
        return type_mapping.get(type(spark_type).__name__, 'string')
    
    def transform_data(self, source_table, target_path, transformation_rules):
        """
        데이터 변환 및 정제 - Curated 계층 생성
        Args:
            source_table: 원본 테이블명
            target_path: 변환된 데이터 저장 경로
            transformation_rules: 변환 규칙 딕셔너리
        """
        try:
            # 원시 데이터 읽기
            source_path = f"s3a://{self.bucket_name}/raw/{source_table}/"
            df = self.spark.read.parquet(source_path)
            
            # 변환 규칙 적용
            transformed_df = df
            
            # 컬럼 이름 정규화
            if 'column_mapping' in transformation_rules:
                for old_name, new_name in transformation_rules['column_mapping'].items():
                    transformed_df = transformed_df.withColumnRenamed(old_name, new_name)
            
            # 데이터 타입 변환
            if 'type_casting' in transformation_rules:
                for column, data_type in transformation_rules['type_casting'].items():
                    transformed_df = transformed_df.withColumn(column, col(column).cast(data_type))
            
            # 필터링 규칙 적용
            if 'filters' in transformation_rules:
                for filter_condition in transformation_rules['filters']:
                    transformed_df = transformed_df.filter(filter_condition)
            
            # 집계 규칙 적용
            if 'aggregations' in transformation_rules:
                group_by_cols = transformation_rules['aggregations']['group_by']
                agg_expressions = transformation_rules['aggregations']['expressions']
                transformed_df = transformed_df.groupBy(*group_by_cols).agg(*agg_expressions)
            
            # Curated 계층에 저장
            curated_path = f"s3a://{self.bucket_name}/curated/{target_path}/"
            transformed_df.write.mode("overwrite").parquet(curated_path)
            
            # 메타데이터 업데이트
            self._update_glue_catalog(f"curated_{target_path}", transformed_df.schema)
            
            self.logger.info(f"Successfully transformed and stored data to {curated_path}")
            
        except Exception as e:
            self.logger.error(f"Error transforming data: {str(e)}")
            raise
    
    def create_data_pipeline(self, pipeline_config):
        """
        데이터 파이프라인 생성 - ETL 워크플로우 자동화
        Args:
            pipeline_config: 파이프라인 설정 정보
        """
        try:
            for step in pipeline_config['steps']:
                if step['type'] == 'ingestion':
                    self.ingest_batch_data(
                        step['source_path'],
                        step['destination_key'],
                        step.get('format', 'parquet')
                    )
                elif step['type'] == 'transformation':
                    self.transform_data(
                        step['source_table'],
                        step['target_path'],
                        step['transformation_rules']
                    )
            
            self.logger.info("Data pipeline executed successfully")
            
        except Exception as e:
            self.logger.error(f"Pipeline execution failed: {str(e)}")
            raise

# 사용 예시
if __name__ == "__main__":
    # 데이터 레이크 매니저 초기화
    dl_manager = DataLakeManager(bucket_name="my-data-lake-bucket")
    
    # 파이프라인 설정 - 고객 데이터 처리 예시
    pipeline_config = {
        'steps': [
            {
                'type': 'ingestion',
                'source_path': '/data/customer_data.parquet',
                'destination_key': 'customer_data',
                'format': 'parquet'
            },
            {
                'type': 'transformation',
                'source_table': 'customer_data',
                'target_path': 'customer_insights',
                'transformation_rules': {
                    'column_mapping': {
                        'cust_id': 'customer_id',
                        'cust_name': 'customer_name'
                    },
                    'type_casting': {
                        'customer_id': 'string',
                        'age': 'integer'
                    },
                    'filters': ['age > 18', 'customer_name is not null'],
                    'aggregations': {
                        'group_by': ['age_group'],
                        'expressions': ['count(*).alias("customer_count")', 
                                      'avg("purchase_amount").alias("avg_purchase")']
                    }
                }
            }
        ]
    }
    
    # 파이프라인 실행
    dl_manager.create_data_pipeline(pipeline_config)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

import boto3
import json
import datetime

# 거래 로그 예시
transaction = {
    "transaction_id": "TX123456",
    "user_id": "U789",
    "amount": 320.50,
    "currency": "USD",
    "timestamp": str(datetime.datetime.now())
}

# S3 저장
s3 = boto3.client('s3')
bucket = "banking-raw-zone"
key = f"transactions/{transaction['transaction_id']}.json"

s3.put_object(Bucket=bucket, Key=key, Body=json.dumps(transaction))
print("Raw transaction saved to S3.")

# Glue Catalog 자동 등록 예시
glue = boto3.client('glue')
glue.create_table(
    DatabaseName='finance_logs',
    TableInput={
        'Name': 'transactions',
        'StorageDescriptor': {
            'Columns': [
                {'Name': 'transaction_id', 'Type': 'string'},
                {'Name': 'user_id', 'Type': 'string'},
                {'Name': 'amount', 'Type': 'double'},
                {'Name': 'currency', 'Type': 'string'},
                {'Name': 'timestamp', 'Type': 'string'}
            ],
            'Location': f"s3://{bucket}/transactions/",
            'InputFormat': 'org.apache.hadoop.mapred.TextInputFormat',
            'OutputFormat': 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat',
            'SerdeInfo': {
                'SerializationLibrary': 'org.openx.data.jsonserde.JsonSerDe',
                'Parameters': {
                    'serialization.format': '1'
                }
            }
        },
        'TableType': 'EXTERNAL_TABLE'
    }
)
print("Glue Catalog 등록 완료.")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# AWS Boto3를 활용한 S3 데이터 적재 및 Athena 쿼리 예시

import boto3

# S3에 데이터 업로드
s3 = boto3.client('s3')
with open('data.csv', 'rb') as data:
    s3.upload_fileobj(data, 'my-datalake-bucket', 'raw/data.csv')

# Glue 카탈로그 등록 및 Athena 쿼리는 AWS 콘솔 또는 SDK로 실행
# Athena에서 쿼리 예시 (SQL)
# SELECT * FROM my_datalake.raw_data WHERE event_type = 'purchase';