5. Filter Interceptor 패턴#

1
# 1단계: 필터 조건 + accumulator 정의
2
@dataclasses.dataclass
3
class FilterWithAccumulator:
4
    name: str
5
    filter: Callable[[Any], bool]
6
    accumulator: Accumulator[int]
7

8
filters_with_accumulators = {
9
    'type': [
10
        FilterWithAccumulator(
11
            'type is null',
12
            lambda device: device['type'] is not None,
13
            spark_context.accumulator(0)
14
        ),
15
        FilterWithAccumulator(
16
            'type is too short (1 char)',
17
            lambda device: len(device['type']) > 1,
18
            spark_context.accumulator(0)
19
        ),
20
    ],
21
}

1
# 2단계: mapInPandas로 필터 평가 + accumulator 증가
2
def filter_null_type(devices_iterator):
3
    def filter_row_with_accumulator(device_row):
4
        for attribute in device_row.keys():
5
            for f in filters_with_accumulators.get(attribute, []):
6
                if not f.filter(device_row):
7
                    f.accumulator.add(1)  # 조건 불만족 → 카운터 증가
8
                    return False
9
        return True
10

11
    for devices_df in devices_iterator:
12
        yield devices_df[devices_df.apply(
13
            lambda device: filter_row_with_accumulator(device), axis=1
14
        ) == True]
15

16
valid_devices = input_dataset.mapInPandas(filter_null_type, schema)
17
valid_devices.write.mode('append').format('delta').save(output_path)

1
# 3단계: 필터링 통계 조회
2
for key, accumulators in filters_with_accumulators.items():
3
    for acc in accumulators:
4
        print(f'{key} // {acc.name}: {acc.accumulator.value}')
5

6
# 출력 예시:
7
# type // type is null: 3
8
# type // type is too short (1 char): 12

1
-- 1단계: 각 필터 조건을 CASE문으로 컬럼화
2
SELECT * FROM (
3
    SELECT
4
        CASE
5
            WHEN (type IS NOT NULL) IS FALSE THEN 'null_type'
6
            WHEN (LEN(type) > 2) IS FALSE THEN 'short_type'
7
            WHEN (full_name IS NOT NULL) IS FALSE THEN 'null_full_name'
8
            WHEN (version IS NOT NULL) IS FALSE THEN 'null_version'
9
            ELSE NULL
10
        END AS status_flag,
11
        type, full_name, version
12
    FROM input
13
);
14

15
-- 2단계: 조건별 필터링 건수 집계
16
SELECT COUNT(*), status_flag FROM input_with_flags
17
WHERE status_flag IS NOT NULL
18
GROUP BY status_flag;
19

20
-- 3단계: 정상 레코드만 추출
21
SELECT type, full_name, version FROM input_with_flags
22
WHERE status_flag IS NULL;