Table API 튜토리얼

Table API 튜토리얼 (Table API Tutorial)

Apache Flink는 배치 및 스트림 처리를 위한 통합된 관계형 API로 Table API를 제공합니다. 쿼리는 무한(unbounded) 실시간 스트림이나 유한(bounded) 배치 데이터에서 동일한 의미론으로 실행되며 동일한 결과를 만듭니다. Flink의 Table API는 데이터 분석, 데이터 파이프라이닝, ETL 애플리케이션에 흔히 사용됩니다.

출처: 문서

본문

무엇을 만들 것인가 (What You'll Build)

이 튜토리얼에서는 계정과 시간 단위로 트랜잭션 금액을 집계하는 지출 보고서(spend report)를 만들 것입니다:

Transactions (generated data) → Flink (Table API aggregation) → Console (results)

다음 내용을 배우게 됩니다:

• Table API 스트리밍 환경 설정
• Table API로 테이블 생성
• Table API 연산으로 연속(continuous) 집계 작성
• 사용자 정의 함수(UDF) 구현
• 집계를 위한 시간 기반 윈도우 사용
• 배치 모드에서 스트리밍 애플리케이션 테스트

사전 요구 사항 (Prerequisites)

이 워크스루는 Java나 Python에 어느 정도 익숙하다고 가정하지만, 다른 프로그래밍 언어에서 왔더라도 따라올 수 있어야 합니다. 또한 SELECT와 GROUP BY 절 같은 기본 관계형 개념에 익숙하다고 가정합니다.

도움이 필요할 때 (Help, I'm Stuck!)

막히면 커뮤니티 지원 리소스를 확인하세요. 특히 Apache Flink 사용자 메일링 리스트는 어떤 Apache 프로젝트보다도 가장 활발한 목록 중 하나로 꾸준히 평가되며, 빠르게 도움을 받을 수 있는 훌륭한 방법입니다.

따라 하는 방법 (How To Follow Along)

따라 하려면 다음을 갖춘 컴퓨터가 필요합니다:

Java

• Java 11, 17 또는 21
• Maven

Python

• Java 11, 17 또는 21
• Python 3.9, 3.10, 3.11 또는 3.12

Java

제공되는 Flink Maven Archetype이 필요한 모든 의존성을 가진 스켈레톤 프로젝트를 빠르게 만들므로, 비즈니스 로직을 채우는 것에만 집중하면 됩니다.

$ mvn archetype:generate \
    -DarchetypeGroupId=org.apache.flink \
    -DarchetypeArtifactId=flink-walkthrough-table-java \
    -DarchetypeVersion=2.3.0 \
    -DgroupId=spendreport \
    -DartifactId=spendreport \
    -Dversion=0.1 \
    -Dpackage=spendreport \
    -DinteractiveMode=false

원하면 groupId, artifactId, package를 편집할 수 있습니다. 위 파라미터로 Maven은 이 튜토리얼을 완료하는 데 필요한 모든 의존성을 가진 프로젝트가 들어 있는 spendreport 폴더를 만듭니다.

프로젝트를 편집기에 가져온 뒤 IDE에서 직접 실행할 수 있는 SpendReport.java 파일을 찾을 수 있습니다.

IDE에서 실행: java.lang.NoClassDefFoundError 예외가 발생하면 클래스패스에 필요한 Flink 의존성이 모두 없기 때문일 가능성이 높습니다.

• IntelliJ IDEA: Run > Edit Configurations > Modify options > "include dependencies with 'Provided' scope" 선택.

Python

Python Table API를 사용하려면 PyFlink를 설치해야 하며, PyFlink는 PyPI에 있으며 pip로 쉽게 설치할 수 있습니다:

$ python -m pip install apache-flink

팁: 프로젝트 의존성을 격리하려면 가상 환경에 PyFlink를 설치하는 것을 권장합니다.

PyFlink가 설치되면 Table API 프로그램을 작성할 spend_report.py 새 파일을 만듭니다.

전체 프로그램 (The Complete Program)

지출 보고서 프로그램의 전체 코드입니다:

Java

public class SpendReport {

    public static void main(String[] args) throws Exception {
        // Create a Table environment for streaming
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment tEnv = TableEnvironment.create(settings);

        // Create the source table
        // The DataGen connector generates an infinite stream of transactions
        tEnv.createTemporaryTable("transactions",
            TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                    .column("accountId", DataTypes.BIGINT())
                    .column("amount", DataTypes.BIGINT())
                    .column("transactionTime", DataTypes.TIMESTAMP(3))
                    .watermark("transactionTime", "transactionTime - INTERVAL '5' SECOND")
                    .build())
                .option("rows-per-second", "100")
                .option("fields.accountId.min", "1")
                .option("fields.accountId.max", "5")
                .option("fields.amount.min", "1")
                .option("fields.amount.max", "1000")
                .build());

        // Read from the source table
        Table transactions = tEnv.from("transactions");

        // Apply the business logic
        Table result = report(transactions);

        // Print the results to the console
        result.execute().print();
    }

    public static Table report(Table transactions) {
        return transactions
            .select(
                $("accountId"),
                $("transactionTime").floor(TimeIntervalUnit.HOUR).as("logTs"),
                $("amount"))
            .groupBy($("accountId"), $("logTs"))
            .select(
                $("accountId"),
                $("logTs"),
                $("amount").sum().as("amount"));
    }
}

Python

from pyflink.table import TableEnvironment, EnvironmentSettings, TableDescriptor, Schema, DataTypes
from pyflink.table.expression import TimeIntervalUnit
from pyflink.table.expressions import col

def main():
    # Create a Table environment for streaming
    settings = EnvironmentSettings.in_streaming_mode()
    t_env = TableEnvironment.create(settings)

    # Write all results to one file for easier viewing
    t_env.get_config().set("parallelism.default", "1")

    # Create the source table
    # The DataGen connector generates an infinite stream of transactions
    t_env.create_temporary_table(
        "transactions",
        TableDescriptor.for_connector("datagen")
            .schema(Schema.new_builder()
                .column("accountId", DataTypes.BIGINT())
                .column("amount", DataTypes.BIGINT())
                .column("transactionTime", DataTypes.TIMESTAMP(3))
                .watermark("transactionTime", "transactionTime - INTERVAL '5' SECOND")
                .build())
            .option("rows-per-second", "100")
            .option("fields.accountId.min", "1")
            .option("fields.accountId.max", "5")
            .option("fields.amount.min", "1")
            .option("fields.amount.max", "1000")
            .build()

    # Read from the source table
    transactions = t_env.from_path("transactions")

    # Apply the business logic
    result = report(transactions)

    # Print the results to the console
    result.execute().print()

def report(transactions):
    return transactions \
        .select(
            col("accountId"),
            col("transactionTime").floor(TimeIntervalUnit.HOUR).alias("logTs"),
            col("amount")) \
        .group_by(col("accountId"), col("logTs")) \
        .select(
            col("accountId"),
            col("logTs"),
            col("amount").sum.alias("amount"))

if __name__ == '__main__':
    main()

코드 분석 (Breaking Down The Code)

실행 환경 (The Execution Environment)

첫 몇 줄은 TableEnvironment를 설정합니다. 테이블 환경은 작업의 속성을 설정하고, 배치 애플리케이션을 작성하는지 스트리밍 애플리케이션을 작성하는지 지정하며, 소스를 만드는 방법입니다. 이 워크스루는 스트리밍 실행을 사용하는 표준 테이블 환경을 만듭니다.

Java

EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
TableEnvironment tEnv = TableEnvironment.create(settings);

Python

settings = EnvironmentSettings.in_streaming_mode()
t_env = TableEnvironment.create(settings)

테이블 생성 (Creating Tables)

다음으로 트랜잭션 데이터를 나타내는 테이블이 생성됩니다. DataGen 커넥터는 무한한 무작위 트랜잭션 스트림을 생성합니다.

Java

TableDescriptor를 사용하면 테이블을 프로그래밍 방식으로 정의할 수 있습니다:

tEnv.createTemporaryTable("transactions",
    TableDescriptor.forConnector("datagen")
        .schema(Schema.newBuilder()
            .column("accountId", DataTypes.BIGINT())
            .column("amount", DataTypes.BIGINT())
            .column("transactionTime", DataTypes.TIMESTAMP(3))
            .watermark("transactionTime", "transactionTime - INTERVAL '5' SECOND")
            .build())
        .option("rows-per-second", "100")
        .option("fields.accountId.min", "1")
        .option("fields.accountId.max", "5")
        .option("fields.amount.min", "1")
        .option("fields.amount.max", "1000")
        .build());

Python

TableDescriptor를 사용하면 테이블을 프로그래밍 방식으로 정의할 수 있습니다:

t_env.create_temporary_table(
    "transactions",
    TableDescriptor.for_connector("datagen")
        .schema(Schema.new_builder()
            .column("accountId", DataTypes.BIGINT())
            .column("amount", DataTypes.BIGINT())
            .column("transactionTime", DataTypes.TIMESTAMP(3))
            .watermark("transactionTime", "transactionTime - INTERVAL '5' SECOND")
            .build())
        .option("rows-per-second", "100")
        .option("fields.accountId.min", "1")
        .option("fields.accountId.max", "5")
        .option("fields.amount.min", "1")
        .option("fields.amount.max", "1000")
        .build()

트랜잭션 테이블은 다음을 가진 신용카드 트랜잭션을 생성합니다:

• accountId: 1과 5 사이의 계정 ID
• amount: 1과 1000 사이의 트랜잭션 금액
• transactionTime: 늦은 데이터 처리를 위한 워터마크가 있는 타임스탬프

쿼리 (The Query)

환경이 설정되고 테이블이 등록되면 첫 번째 애플리케이션을 만들 준비가 된 것입니다. TableEnvironment에서 입력 테이블을 from으로 읽고 Table API 연산을 적용할 수 있습니다. report 함수가 비즈니스 로직을 구현하는 곳입니다.

Java

Table transactions = tEnv.from("transactions");
Table result = report(transactions);
result.execute().print();

Python

transactions = t_env.from_path("transactions")
result = report(transactions)
result.execute().print()

보고서 구현 (Implementing the Report)

이제 작업의 골격이 설정되었으니 비즈니스 로직을 추가할 준비가 되었습니다. 목표는 하루의 각 시간에 대해 계정별 총 지출을 보여주는 보고서를 만드는 것입니다. 이는 타임스탬프 컬럼을 밀리초에서 시간 단위로 반올림(내림)해야 함을 의미합니다.

Flink는 순수 SQL 또는 Table API를 사용한 관계형 애플리케이션 개발을 지원합니다. Table API는 SQL에서 영감을 받은 유창한(fluent) DSL로 Java나 Python으로 작성할 수 있고 강력한 IDE 통합을 지원합니다. SQL 쿼리처럼 Table 프로그램은 필요한 필드를 선택하고 키로 그룹화할 수 있습니다. 이러한 기능은 floor와 sum 같은 내장 함수와 함께 이 보고서를 작성할 수 있게 해줍니다.

Java

public static Table report(Table transactions) {
    return transactions
        .select(
            $("accountId"),
            $("transactionTime").floor(TimeIntervalUnit.HOUR).as("logTs"),
            $("amount"))
        .groupBy($("accountId"), $("logTs"))
        .select(
            $("accountId"),
            $("logTs"),
            $("amount").sum().as("amount"));
}

Python

def report(transactions):
    return transactions \
        .select(
            col("accountId"),
            col("transactionTime").floor(TimeIntervalUnit.HOUR).alias("logTs"),
            col("amount")) \
        .group_by(col("accountId"), col("logTs")) \
        .select(
            col("accountId"),
            col("logTs"),
            col("amount").sum.alias("amount"))

테스트 (Testing)

Java

프로젝트에는 정적 데이터로 배치 모드를 사용해 보고서의 로직을 검증하는 SpendReportTest 테스트 클래스가 포함되어 있습니다.

EnvironmentSettings settings = EnvironmentSettings.inBatchMode();
TableEnvironment tEnv = TableEnvironment.create(settings);

// Create test data using fromValues
Table transactions = tEnv.fromValues(
    DataTypes.ROW(
        DataTypes.FIELD("accountId", DataTypes.BIGINT()),
        DataTypes.FIELD("amount", DataTypes.BIGINT()),
        DataTypes.FIELD("transactionTime", DataTypes.TIMESTAMP(3))
    ),
    Row.of(1L, 188L, LocalDateTime.of(2024, 1, 1, 9, 0, 0)),
    Row.of(1L, 374L, LocalDateTime.of(2024, 1, 1, 9, 30, 0)),
    // ... more test data
);

Python

배치 모드로 전환하고 정적 테스트 데이터를 사용해 report 함수를 테스트할 수 있습니다. 별도의 테스트 파일(예: test_spend_report.py)을 만듭니다:

from datetime import datetime
from pyflink.table import TableEnvironment, EnvironmentSettings, DataTypes

from spend_report import report

def test_report():
    settings = EnvironmentSettings.in_batch_mode()
    t_env = TableEnvironment.create(settings)

    # Create test data using from_elements
    transactions = t_env.from_elements(
        [
            (1, 188, datetime(2024, 1, 1, 9, 0, 0)),
            (1, 374, datetime(2024, 1, 1, 9, 30, 0)),
            (2, 200, datetime(2024, 1, 1, 9, 15, 0)),
        ],
        DataTypes.ROW([
            DataTypes.FIELD("accountId", DataTypes.BIGINT()),
            DataTypes.FIELD("amount", DataTypes.BIGINT()),
            DataTypes.FIELD("transactionTime", DataTypes.TIMESTAMP(3))
        ])
    )

    # Test the report function
    result = report(transactions)

    # Collect results and verify
    rows = [row for row in result.execute().collect()]
    assert len(rows) == 2  # Two accounts

if __name__ == '__main__':
    test_report()
    print("All tests passed!")

python test_spend_report.py 또는 pytest test_spend_report.py로 실행합니다.

Flink의 고유한 속성 중 하나는 배치와 스트리밍에서 일관된 의미론을 제공한다는 것입니다. 이는 정적 데이터셋에서 배치 모드로 애플리케이션을 개발하고 테스트한 뒤, 스트리밍 애플리케이션으로 프로덕션에 배포할 수 있음을 의미합니다.

사용자 정의 함수 (User Defined Functions)

Flink에는 여러 내장 함수가 있으며, 때로는 사용자 정의 함수로 확장해야 합니다. floor가 미리 정의되어 있지 않다면 직접 구현할 수 있습니다.

Java

import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;

import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ScalarFunction;

public class MyFloor extends ScalarFunction {

    public @DataTypeHint("TIMESTAMP(3)") LocalDateTime eval(
        @DataTypeHint("TIMESTAMP(3)") LocalDateTime timestamp) {

        return timestamp.truncatedTo(ChronoUnit.HOURS);
    }
}

그런 다음 애플리케이션에 빠르게 통합합니다:

public static Table report(Table transactions) {
    return transactions
        .select(
            $("accountId"),
            call(MyFloor.class, $("transactionTime")).as("logTs"),
            $("amount"))
        .groupBy($("accountId"), $("logTs"))
        .select(
            $("accountId"),
            $("logTs"),
            $("amount").sum().as("amount"));
}

Python

from pyflink.table.udf import udf
from datetime import datetime

@udf(result_type=DataTypes.TIMESTAMP(3))
def my_floor(timestamp: datetime) -> datetime:
    return timestamp.replace(minute=0, second=0, microsecond=0)

그런 다음 애플리케이션에 통합합니다:

def report(transactions):
    return transactions \
        .select(
            col("accountId"),
            my_floor(col("transactionTime")).alias("logTs"),
            col("amount")) \
        .group_by(col("accountId"), col("logTs")) \
        .select(
            col("accountId"),
            col("logTs"),
            col("amount").sum.alias("amount"))

이 쿼리는 transactions 테이블의 모든 레코드를 소비하고, 보고서를 계산하며, 결과를 효율적이고 확장 가능한 방식으로 출력합니다. 이 구현으로 테스트를 실행하면 통과합니다.

프로세스 테이블 함수 (Process Table Functions, Java 전용)

더 고급의 행 단위 처리를 위해 Flink는 프로세스 테이블 함수(PTF)를 제공합니다. PTF는 테이블의 각 행을 변환할 수 있고 상태(state)와 타이머(timer) 같은 강력한 기능에 접근할 수 있습니다. 다음은 고액 트랜잭션을 필터링하고 포맷하는 간단한 무상태 예시입니다:

import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.ArgumentTrait;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.types.Row;

public class HighValueAlerts extends ProcessTableFunction<String> {

    private static final long HIGH_VALUE_THRESHOLD = 500;

    public void eval(@ArgumentHint(ArgumentTrait.ROW_SEMANTIC_TABLE) Row transaction) {
        Long amount = transaction.getFieldAs("amount");
        if (amount > HIGH_VALUE_THRESHOLD) {
            Long accountId = transaction.getFieldAs("accountId");
            collect("Alert: Account " + accountId + " made a high-value transaction of " + amount);
        }
    }
}

main() 메서드를 수정해 이 PTF를 시도할 수 있습니다. 기존 report() 호출을 대체하거나 옆에 추가합니다:

// Instead of (or in addition to) the aggregation report:
// Table result = report(transactions);
// result.execute().print();

// Try the PTF to see high-value transaction alerts:
Table alerts = transactions.process(HighValueAlerts.class);
alerts.execute().print();

이것은 500을 초과하는 트랜잭션에 대해서만 알림을 출력합니다. PTF는 상태와 타이머와 결합하면 복잡한 이벤트 기반 로직을 구현할 때 훨씬 더 강력해집니다 — 더 고급 예시는 PTF 문서를 참고하세요.

참고: 프로세스 테이블 함수는 현재 Java에서만 사용할 수 있습니다. Python에서는 유사한 행 단위 처리를 위해 사용자 정의 테이블 함수(UDTF)를 사용할 수 있습니다.

윈도우 추가 (Adding Windows)

시간에 따라 데이터를 그룹화하는 것은 데이터 처리의 전형적인 연산이며, 특히 무한 스트림을 다룰 때 그렇습니다. 시간에 따른 그룹화를 윈도우(window)라고 하며 Flink는 유연한 윈도우 의미론을 제공합니다. 윈도우의 가장 기본적인 유형은 Tumble 윈도우로, 고정된 크기를 가지며 버킷이 겹치지 않습니다.

report() 함수를 floor() 대신 윈도우를 사용하도록 수정해 보세요:

Java

public static Table report(Table transactions) {
    return transactions
        .window(Tumble.over(lit(10).seconds()).on($("transactionTime")).as("logTs"))
        .groupBy($("accountId"), $("logTs"))
        .select(
            $("accountId"),
            $("logTs").start().as("logTs"),
            $("amount").sum().as("amount"));
}

Python

from pyflink.table.expressions import col, lit
from pyflink.table.window import Tumble

def report(transactions):
    return transactions \
        .window(Tumble.over(lit(10).seconds).on(col("transactionTime")).alias("logTs")) \
        .group_by(col("accountId"), col("logTs")) \
        .select(
            col("accountId"),
            col("logTs").start.alias("logTs"),
            col("amount").sum.alias("amount"))

이것은 타임스탬프 컬럼을 기준으로 10초 텀블링 윈도우를 사용하도록 애플리케이션을 정의합니다. 따라서 타임스탬프가 2024-01-01 01:23:47인 행은 2024-01-01 01:23:40 윈도우에 들어갑니다.

시간 기반 집계는 시간이 다른 속성과 달리 연속 스트리밍 애플리케이션에서 일반적으로 앞으로 진행되기 때문에 고유합니다. floor와 여러분의 UDF와 달리 윈도우 함수는 내장(intrinsics)이므로 런타임이 추가 최적화를 적용할 수 있습니다. 배치 컨텍스트에서 윈도우는 타임스탬프 속성으로 레코드를 그룹화하는 편리한 API를 제공합니다.

이 변경 후 애플리케이션을 실행하면 10초마다 윈도우 결과가 출력되는 것을 볼 수 있습니다.

애플리케이션 실행 (Running the Application)

완전히 기능하는 상태 저장 분산 스트리밍 애플리케이션이 완성되었습니다! 쿼리는 트랜잭션을 계속 생성하고, 시간별 지출을 계산하며, 준비되는 대로 결과를 내보냅니다. 입력이 무한하므로 쿼리는 수동으로 중지할 때까지 계속 실행됩니다.

Java

IDE에서 SpendReport 클래스를 실행하면 콘솔에 스트리밍 결과가 출력되는 것을 볼 수 있습니다.

Python

명령줄에서 프로그램을 실행합니다:

$ python spend_report.py

이 명령은 로컬 미니 클러스터에서 Python Table API 프로그램을 빌드하고 실행합니다. Python Table API 프로그램을 원격 클러스터에 제출할 수도 있습니다. 자세한 내용은 Job Submission Examples를 참고하세요.

다음 단계 (Next Steps)

튜토리얼 완료를 축하합니다! 학습을 계속하기 위한 몇 가지 방법입니다:

Table API에 대해 더 배우기 (Learn More About the Table API)

• Table API 개요: 완전한 Table API 참조
• 사용자 정의 함수: 파이프라인을 위한 커스텀 함수 생성
• 스트리밍 개념: 동적 테이블, 시간 속성 등 이해

다른 튜토리얼 탐색 (Explore Other Tutorials)

• Flink SQL 튜토리얼: 코딩 없이 대화형 SQL 쿼리
• DataStream API 튜토리얼: DataStream API로 상태 저장 스트리밍 애플리케이션 구축
• Flink Operations Playground: Flink 클러스터 운영 배우기

프로덕션 배포 (Production Deployment)

• 배포 개요: 프로덕션에 Flink 배포
• 커넥터: Kafka, 데이터베이스, 파일 시스템 등에 연결

더 알아보기 (Learn more)

• Table API Overview
• Flink SQL Tutorial
• DataStream API Tutorial
• Table API Concepts