State Processor API

State Processor API

Apache Flink의 State Processor API는 BATCH 실행 하에서 Flink의 DataStream API와 Table API를 사용해 savepoint과 checkpoint를 읽고, 쓰고, 수정하는 강력한 기능을 제공합니다. DataStream과 Table API의 상호 운용성 덕분에 관계형 Table API 또는 SQL 쿼리를 사용해 상태 데이터를 분석하고 처리할 수도 있습니다.

예를 들어 실행 중인 스트림 처리 애플리케이션의 savepoint을 가져와 DataStream 배치 프로그램으로 분석하여 애플리케이션이 올바르게 동작하는지 검증할 수 있습니다. 또는 어떤 저장소에서 배치 데이터를 읽어 전처리한 후 그 결과를 스트리밍 애플리케이션의 상태를 부트스트랩하는 데 사용하는 savepoint에 쓸 수 있습니다. 또한 일관되지 않은 상태 항목을 수정하는 것도 가능합니다. 마지막으로 State Processor API는 상태 기반 애플리케이션을 진화시키는 많은 방법을 열어줍니다. 이전에는 시작 후 애플리케이션의 모든 상태를 잃지 않고는 변경할 수 없는 매개변수와 설계 선택에 의해 막혀 있었습니다. 예를 들어 이제 상태의 데이터 타입을 임의로 수정하고, 연산자의 최대 병렬도를 조정하고, 연산자 상태를 분할하거나 병합하고, 연산자 UID를 재할당하는 등을 할 수 있습니다.

State processor API를 시작하려면 애플리케이션에 다음 라이브러리를 포함하세요.

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-state-processor-api</artifactId>
    <version>2.3.0</version>
</dependency>

출처: 문서

본문

애플리케이션 상태를 DataSet에 매핑

State Processor API는 스트리밍 애플리케이션의 상태를 별도로 처리할 수 있는 하나 이상의 데이터 집합에 매핑합니다. API를 사용하려면 이 매핑이 어떻게 작동하는지 이해해야 합니다.

먼저 상태 기반 Flink 작업이 어떻게 생겼는지 살펴봅시다. Flink 작업은 연산자로 구성됩니다. 일반적으로 하나 이상의 소스 연산자, 실제 처리를 위한 몇 개의 연산자, 하나 이상의 싱크 연산자입니다. 각 연산자는 하나 이상의 태스크에서 병렬로 실행되며 다양한 유형의 상태로 작업할 수 있습니다. 연산자는 연산자의 태스크로 범위가 지정된 목록으로 구성된 *"operator states"*를 0개, 1개 또는 여러 개 가질 수 있습니다. 연산자가 keyed 스트림에 적용되면 처리된 각 레코드에서 추출된 키로 범위가 지정되는 *"keyed states"*를 0개, 1개 또는 여러 개 가질 수도 있습니다. Keyed state를 분산 키-값 맵으로 생각할 수 있습니다.

다음 그림은 "Src", "Proc", "Snk"라는 세 연산자로 구성된 애플리케이션 "MyApp"을 보여줍니다. Src는 하나의 operator state(os1)를, Proc는 하나의 operator state(os2)와 두 개의 keyed state(ks1, ks2)를 가지며 Snk는 무상태입니다.

MyApp의 savepoint 또는 checkpoint는 각 태스크의 상태를 복원할 수 있는 방식으로 구성된 모든 상태의 데이터로 구성됩니다. 배치 작업으로 savepoint(또는 checkpoint)의 데이터를 처리할 때 개별 태스크 상태의 데이터를 데이터 집합이나 테이블에 매핑하는 정신적 모델이 필요합니다. 실제로 savepoint을 데이터베이스로 생각할 수 있습니다. 각 연산자(UID로 식별)는 네임스페이스를 나타냅니다. 연산자의 각 operator state는 모든 태스크의 상태 데이터를 보유하는 단일 열을 가진 네임스페이스의 전용 테이블에 매핑됩니다. 연산자의 모든 keyed state는 키를 위한 열 하나와 각 keyed state를 위한 열 하나로 구성된 단일 테이블에 매핑됩니다. 다음 그림은 MyApp의 savepoint이 데이터베이스에 어떻게 매핑되는지 보여줍니다.

그림은 Src의 operator state 값이 한 열과 다섯 행(각 행은 Src의 모든 병렬 태스크에 걸친 목록 항목 중 하나)을 가진 테이블에 매핑되는 방법을 보여줍니다. "Proc" 연산자의 operator state os2도 유사하게 개별 테이블에 매핑됩니다. Keyed state ks1과 ks2는 키, ks1, ks2를 위한 세 열로 구성된 단일 테이블로 결합됩니다. keyed 테이블은 두 keyed state의 각 고유 키에 대해 한 행을 보유합니다. "Snk" 연산자는 상태가 없으므로 그 네임스페이스는 비어 있습니다.

연산자 식별

State Processor API는 OperatorIdentifier#forUid/forUidHash를 통해 UID 또는 UID 해시로 연산자를 식별할 수 있게 합니다. 해시는 UID의 사용이 불가능할 때만 사용해야 합니다. 예를 들어 savepoint을 만든 애플리케이션이 이를 지정하지 않았거나 UID를 알 수 없는 경우입니다.

DataStream API

상태 읽기

상태 읽기는 유효한 savepoint 또는 checkpoint의 경로와 데이터를 복원하는 데 사용할 StateBackend를 지정하는 것으로 시작합니다. 상태 복원의 호환성 보장은 DataStream 애플리케이션을 복원할 때와 동일합니다.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
SavepointReader savepoint = SavepointReader.read(env, "hdfs://path/", new HashMapStateBackend());

Operator State

Operator state는 Flink의 모든 비-keyed 상태입니다. 여기에는 애플리케이션 내의 CheckpointedFunction 또는 BroadcastState의 사용이 포함되지만 이에 국한되지는 않습니다. Operator state를 읽을 때 사용자는 연산자 uid, 상태 이름, 타입 정보를 지정합니다.

Operator List State

getListState를 사용하는 CheckpointedFunction에 저장된 operator state는 SavepointReader#readListState로 읽을 수 있습니다. 상태 이름과 타입 정보는 DataStream 애플리케이션에서 이 상태를 선언한 ListStateDescriptor를 정의하는 데 사용된 것과 일치해야 합니다.

DataStream<Integer> listState  = savepoint.readListState<>(
    OperatorIdentifier.forUid("my-uid"),
    "list-state",
    Types.INT);

Operator Union List State

getUnionListState를 사용하는 CheckpointedFunction에 저장된 operator state는 SavepointReader#readUnionState로 읽을 수 있습니다. 상태 이름과 타입 정보는 DataStream 애플리케이션에서 이 상태를 선언한 ListStateDescriptor를 정의하는 데 사용된 것과 일치해야 합니다. 프레임워크는 병렬도 1의 DataStream 복원과 동등한 상태의 단일 복사본을 반환합니다.

DataStream<Integer> listState  = savepoint.readUnionState<>(
    OperatorIdentifier.forUid("my-uid"),
    "union-state",
    Types.INT);

Broadcast State

BroadcastState는 SavepointReader#readBroadcastState로 읽을 수 있습니다. 상태 이름과 타입 정보는 DataStream 애플리케이션에서 이 상태를 선언한 MapStateDescriptor를 정의하는 데 사용된 것과 일치해야 합니다. 프레임워크는 병렬도 1의 DataStream 복원과 동등한 상태의 단일 복사본을 반환합니다.

DataStream<Tuple2<Integer, Integer>> broadcastState = savepoint.readBroadcastState<>(
    OperatorIdentifier.forUid("my-uid"),
    "broadcast-state",
    Types.INT,
    Types.INT);

사용자 지정 직렬화기 사용

각 operator state reader는 상태를 쓴 StateDescriptor를 정의하는 데 사용된 경우 사용자 지정 TypeSerializer 사용을 지원합니다.

DataStream<Integer> listState = savepoint.readListState<>(
    OperatorIdentifier.forUid("uid"),
    "list-state", 
    Types.INT,
    new MyCustomIntSerializer());

Keyed State

Keyed state(파티션된 상태라고도 함)는 키에 상대적으로 파티션된 모든 상태입니다. Keyed state를 읽을 때 사용자는 연산자 id와 KeyedStateReaderFunction<KeyType, OutputType>를 지정합니다.

KeyedStateReaderFunction은 임의의 열과 ListState, MapState, AggregatingState 같은 복잡한 상태 유형을 읽을 수 있게 합니다. 즉 연산자가 다음과 같은 상태 기반 process function을 포함한다면:

public class StatefulFunctionWithTime extends KeyedProcessFunction<Integer, Integer, Void> {
 
   ValueState<Integer> state;
 
   ListState<Long> updateTimes;

   @Override
   public void open(OpenContext openContext) {
      ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("state", Types.INT);
      state = getRuntimeContext().getState(stateDescriptor);

      ListStateDescriptor<Long> updateDescriptor = new ListStateDescriptor<>("times", Types.LONG);
      updateTimes = getRuntimeContext().getListState(updateDescriptor);
   }
 
   @Override
   public void processElement(Integer value, Context ctx, Collector<Void> out) throws Exception {
      state.update(value + 1);
      updateTimes.add(System.currentTimeMillis());
   }
}

출력 타입과 해당 KeyedStateReaderFunction을 정의하여 읽을 수 있습니다.

DataStream<KeyedState> keyedState = savepoint.readKeyedState(OperatorIdentifier.forUid("my-uid"), new ReaderFunction());

public class KeyedState {
  public int key;

  public int value;

  public List<Long> times;
}
 
public class ReaderFunction extends KeyedStateReaderFunction<Integer, KeyedState> {

  ValueState<Integer> state;
 
  ListState<Long> updateTimes;

  @Override
  public void open(OpenContext openContext) {
    ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("state", Types.INT);
    state = getRuntimeContext().getState(stateDescriptor);

    ListStateDescriptor<Long> updateDescriptor = new ListStateDescriptor<>("times", Types.LONG);
    updateTimes = getRuntimeContext().getListState(updateDescriptor);
  }
 
  @Override
  public void readKey(
    Integer key,
    Context ctx,
    Collector<KeyedState> out) throws Exception {
        
    KeyedState data = new KeyedState();
    data.key    = key;
    data.value  = state.value();
    data.times  = StreamSupport
      .stream(updateTimes.get().spliterator(), false)
      .collect(Collectors.toList());

    out.collect(data);
  }
}

등록된 상태 값 읽기와 함께 각 키는 등록된 이벤트 시간 및 처리 시간 타이머 같은 메타데이터가 있는 Context에 접근할 수 있습니다.

참고: KeyedStateReaderFunction을 사용할 때 모든 상태 기술자는 open 내부에서 적극적으로(eagerly) 등록되어야 합니다. RuntimeContext#get*State를 호출하려는 시도는 RuntimeException을 초래합니다.

Window State

State processor api는 window 연산자에서 상태 읽기를 지원합니다. Window state를 읽을 때 사용자는 연산자 id, window assigner, 집계 타입을 지정합니다.

또한 WindowFunction 또는 ProcessWindowFunction과 유사하게 각 읽기에 추가 정보를 풍부하게 하기 위해 WindowReaderFunction을 지정할 수 있습니다.

사용자당 분당 클릭 수를 계산하는 DataStream 애플리케이션을 가정해 봅시다.

class Click {
    public String userId;

    public LocalDateTime time;    
}

class ClickCounter implements AggregateFunction<Click, Integer, Integer> {

	@Override
	public Integer createAccumulator() {
		return 0;
	}

	@Override
	public Integer add(Click value, Integer accumulator) {
		return 1 + accumulator;
	}

	@Override
	public Integer getResult(Integer accumulator) {
		return accumulator;
	}

	@Override
	public Integer merge(Integer a, Integer b) {
		return a + b;
	}
}

DataStream<Click> clicks = ...;

clicks
    .keyBy(click -> click.userId)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .aggregate(new ClickCounter())
    .uid("click-window")
    .addSink(new Sink());

이 상태는 아래 코드로 읽을 수 있습니다.

class ClickState {
    
    public String userId;

    public int count;

    public TimeWindow window;

    public Set<Long> triggerTimers;
}

class ClickReader extends WindowReaderFunction<Integer, ClickState, String, TimeWindow> { 

	@Override
	public void readWindow(
            String key,
            Context<TimeWindow> context,
            Iterable<Integer> elements,
            Collector<ClickState> out) {
		ClickState state = new ClickState();
		state.userId = key;
		state.count = elements.iterator().next();
		state.window = context.window();
		state.triggerTimers = context.registeredEventTimeTimers();
		
		out.collect(state);
	}
}

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
SavepointReader savepoint = SavepointReader.read(env, "hdfs://checkpoint-dir", new HashMapStateBackend());

savepoint
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .aggregate("click-window", new ClickCounter(), new ClickReader(), Types.String, Types.INT, Types.INT)
    .print();

또한 CountTrigger 또는 사용자 지정 trigger의 trigger state는 WindowReaderFunction 내부의 Context#triggerState 메서드로 읽을 수 있습니다.

새 Savepoint 쓰기

Savepoint도 쓸 수 있으며, 이를 통해 과거 데이터를 기반으로 상태를 부트스트랩하는 것 같은 사용 사례가 가능합니다. 각 savepoint은 개별 연산자의 상태를 정의하는 하나 이상의 StateBootstrapTransformation(아래 설명)으로 구성됩니다.

SavepointWriter를 사용할 때 애플리케이션은 BATCH 실행으로 실행되어야 합니다.

int maxParallelism = 128;

SavepointWriter
    .newSavepoint(env, new HashMapStateBackend(), maxParallelism)
    .withOperator(OperatorIdentifier.forUid("uid1"), transformation1)
    .withOperator(OperatorIdentifier.forUid("uid2"), transformation2)
    .write(savepointPath);

각 연산자와 연관된 UID는 DataStream 애플리케이션의 연산자에 할당된 UID와 일대일로 일치해야 합니다. 이것이 Flink가 어떤 상태가 어떤 연산자에 매핑되는지 알게 하는 방법입니다.

Operator State

CheckpointedFunction을 사용하는 단순 operator state는 StateBootstrapFunction으로 생성할 수 있습니다.

public class SimpleBootstrapFunction extends StateBootstrapFunction<Integer> {

    private ListState<Integer> state;

    @Override
    public void processElement(Integer value, Context ctx) throws Exception {
        state.add(value);
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
    }
	
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        state = context.getOperatorState().getListState(new ListStateDescriptor<>("state", Types.INT));
    }
}

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Integer> data = env.fromElements(1, 2, 3);

StateBootstrapTransformation transformation = OperatorTransformation
    .bootstrapWith(data)
    .transform(new SimpleBootstrapFunction());

Broadcast State

BroadcastState는 BroadcastStateBootstrapFunction으로 쓸 수 있습니다. DataStream API의 broadcast state와 유사하게 전체 상태는 메모리에 맞아야 합니다.

public class CurrencyRate {
    public String currency;

    public Double rate;
}

public class CurrencyBootstrapFunction extends BroadcastStateBootstrapFunction<CurrencyRate> {

    public static final MapStateDescriptor<String, Double> descriptor = 
        new MapStateDescriptor<>("currency-rates", Types.STRING, Types.DOUBLE);

    @Override
    public void processElement(CurrencyRate value, Context ctx) throws Exception {
        ctx.getBroadcastState(descriptor).put(value.currency, value.rate);
    }
}

DataStream<CurrencyRate> currencyDataSet = env.fromCollection(
    new CurrencyRate("USD", 1.0), new CurrencyRate("EUR", 1.3));

StateBootstrapTransformation<CurrencyRate> broadcastTransformation = OperatorTransformation
    .bootstrapWith(currencyDataSet)
    .transform(new CurrencyBootstrapFunction());

Keyed State

ProcessFunction 및 기타 RichFunction 타입을 위한 keyed state는 KeyedStateBootstrapFunction으로 쓸 수 있습니다.

public class Account {
    public int id;

    public double amount;	

    public long timestamp;
}
 
public class AccountBootstrapper extends KeyedStateBootstrapFunction<Integer, Account> {
    ValueState<Double> state;

    @Override
    public void open(OpenContext openContext) {
        ValueStateDescriptor<Double> descriptor = new ValueStateDescriptor<>("total",Types.DOUBLE);
        state = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(Account value, Context ctx) throws Exception {
        state.update(value.amount);
    }
}
 
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<Account> accountDataSet = env.fromCollection(accounts);

StateBootstrapTransformation<Account> transformation = OperatorTransformation
    .bootstrapWith(accountDataSet)
    .keyBy(acc -> acc.id)
    .transform(new AccountBootstrapper());

KeyedStateBootstrapFunction은 이벤트 시간 및 처리 시간 타이머 설정을 지원합니다. 타이머는 부트스트랩 함수 내부에서 실행되지 않으며 DataStream 애플리케이션에서 복원된 후에만 활성화됩니다. 처리 시간 타이머가 설정되었지만 그 시간이 지난 후에 상태가 복원되면 타이머는 시작 즉시 실행됩니다.

주의: 부트스트랩 함수가 타이머를 만들면 상태는 process 유형 함수 중 하나로만 복원할 수 있습니다.

Window State

State processor api는 window 연산자의 상태 쓰기를 지원합니다. Window state를 쓸 때 사용자는 연산자 id, window assigner, evictor, 선택적 trigger, 집계 타입을 지정합니다. 부트스트랩 변환의 구성이 DataStream window의 구성과 일치하는 것이 중요합니다.

public class Account {
    public int id;

    public double amount;	

    public long timestamp;
}

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<Account> accountDataSet = env.fromCollection(accounts);

StateBootstrapTransformation<Account> transformation = OperatorTransformation
    .bootstrapWith(accountDataSet)
    .keyBy(acc -> acc.id)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(5)))
    .reduce((left, right) -> left + right);

Savepoint 수정

처음부터 savepoint을 만드는 것 외에도 기존 작업에 단일 새 연산자를 부트스트랩하는 경우처럼 기존 savepoint을 기반으로 할 수 있습니다.

SavepointWriter
    .fromExistingSavepoint(env, oldPath, new HashMapStateBackend())
    .withOperator(OperatorIdentifier.forUid("uid"), transformation)
    .write(newPath);

UID(해시) 변경

SavepointWriter#changeOperatorIdenfifier는 연산자의 UID 또는 UID hash를 수정하는 데 사용할 수 있습니다.

UID가 명시적으로 설정되지 않았다면(따라서 자동 생성되어 사실상 알 수 없음) UID hash를 안다면(예: 로그 파싱으로) UID를 할당할 수 있습니다.

savepointWriter
    .changeOperatorIdentifier(
        OperatorIdentifier.forUidHash("2feb7f8bcc404c3ac8a981959780bd78"),
        OperatorIdentifier.forUid("new-uid"))
    ...

하나의 UID를 다른 것으로 교체할 수도 있습니다.

savepointWriter
    .changeOperatorIdentifier(
        OperatorIdentifier.forUid("old-uid"),
        OperatorIdentifier.forUid("new-uid"))
    ...

Table API

시작하기

Table API로 상태를 조사하기 전에 Flink SQL 가이드라인을 검토해야 합니다.

중요한 참고: State Table API는 keyed state만 지원합니다.

메타데이터

다음 SQL 테이블 함수는 사용자가 savepoint과 checkpoint의 메타데이터를 다음과 같은 방식으로 읽을 수 있게 합니다.

LOAD MODULE state;
SELECT * FROM savepoint_metadata('/root/dir/of/checkpoint-data/chk-1');

새 테이블 함수는 다음과 같은 고정 스키마를 가진 테이블을 만듭니다.

Key	Data type	Description
checkpoint-id	BIGINT NOT NULL	Checkpoint ID.
operator-name	STRING	Operator Name.
operator-uid	STRING	Operator UID.
operator-uid-hash	STRING NOT NULL	Operator UID hash.
operator-parallelism	INT NOT NULL	Parallelism of the operator.
operator-max-parallelism	INT NOT NULL	Maximum parallelism of the operator.
operator-subtask-state-count	INT NOT NULL	Number of operator subtask states. It represents the state partition count divided by the operator's parallelism and might be 0 if the state is not partitioned (for example broadcast source).
operator-coordinator-state-size-in-bytes	BIGINT NOT NULL	The operator's coordinator state size in bytes, or zero if no coordinator state.
operator-total-size-in-bytes	BIGINT NOT NULL	Total operator state size in bytes.

Keyed State

Keyed state(파티션된 상태라고도 함)는 키에 상대적으로 파티션된 모든 상태입니다.

SQL 커넥터는 임의의 열을 ValueState로 읽고 ListState, MapState 같은 복잡한 상태 유형을 읽을 수 있게 합니다. 즉 연산자가 다음과 같은 상태 기반 process function을 포함한다면:

eventStream
  .keyBy(e -> (Integer)e.key)
  .process(new StatefulFunction())
  .uid("my-uid");

...

public class Account {
    private Integer id;
    public Double amount;

    public Integer geId() {
        return id;
    }

    public void setId(Integer id) {
        this.id = id;
    }
}

@org.apache.avro.specific.AvroGenerated
public class AvroRecord ... {
    // Generated record which contains at least the following field: long longData
}

public class StatefulFunction extends KeyedProcessFunction<Integer, Integer, Void> {
  private ValueState<Integer> myValueState;
  private ValueState<Account> myAccountValueState;
  private ListState<Integer> myListState;
  private MapState<Integer, Integer> myMapState;
  private ValueState<AvroRecord> myAvroState;

  @Override
  public void open(OpenContext openContext) {
    myValueState = getRuntimeContext().getState(new ValueStateDescriptor<>("MyValueState", Integer.class));
    myAccountValueState = getRuntimeContext().getState(new ValueStateDescriptor<>("MyAccountValueState", Account.class));
    myValueState = getRuntimeContext().getListState(new ListStateDescriptor<>("MyListState", Integer.class));
    myMapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("MyMapState", Integer.class, Integer.class));
    myAvroState = getRuntimeContext().getMapState(new ValueStateDescriptor<>("MyAvroState", new AvroTypeInfo<>(AvroRecord.class)));
  }
  ...
}

그러면 다음 SQL 문으로 생성된 테이블을 쿼리하여 읽을 수 있습니다.

CREATE TABLE state_table (
  k INTEGER,
  MyValueState INTEGER,
  MyAccountValueState ROW<id INTEGER, amount DOUBLE>,
  MyListState ARRAY<INTEGER>,
  MyMapState MAP<INTEGER, INTEGER>,
  MyAvroState ROW<longData bigint>,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'connector' = 'savepoint',
  'state.backend.type' = 'rocksdb',
  'state.path' = '/root/dir/of/checkpoint-data/chk-1',
  'operator.uid' = 'my-uid'
);

커넥터 옵션

일반 옵션

Option	Required	Default	Type	Description
connector	required	(none)	String	사용할 커넥터를 지정합니다. 여기서는 'savepoint'여야 합니다.
state.backend.type	optional	(none)	Enum Possible values: hashmap, rocksdb	상태 읽기에 사용해야 하는 state backend를 정의합니다. 이 값은 savepoint 또는 checkpoint를 만든 Flink 작업에 정의된 값과 일치해야 합니다. 제공되지 않으면 flink 구성의 state.backend.type으로 폴백합니다.
state.path	required	(none)	String	상태 읽기에 사용해야 하는 상태 경로를 정의합니다. Flink가 지원하는 모든 파일 시스템을 여기 사용할 수 있습니다.
operator.uid	optional	(none)	String	상태 읽기에 사용해야 하는 연산자 UID를 정의합니다(operator.uid.hash와 함께 사용할 수 없음). operator.uid 또는 operator.uid.hash 중 하나가 지정되어야 합니다.
operator.uid.hash	optional	(none)	String	상태 읽기에 사용해야 하는 연산자 UID 해시를 정의합니다(operator.uid와 함께 사용할 수 없음). operator.uid 또는 operator.uid.hash 중 하나가 지정되어야 합니다.

열 '#'의 커넥터 옵션

Option	Required	Default	Type	Description
fields.#.state-name	optional	(none)	String	상태 읽기에 사용해야 하는 상태 이름을 재정의합니다. 상태 이름이 SQL 열 이름과 호환되지 않는 문자를 포함할 때 유용할 수 있습니다.
fields.#.state-type	optional	(none)	Enum Possible values: list, map, value	상태 읽기에 사용해야 하는 상태 유형을 정의합니다. value, list, map을 포함합니다. 제공되지 않으면 SQL 타입에서 추론하려고 시도합니다(ARRAY=list, MAP=map, 그 외 모두=value).
fields.#.key-class	optional	(none)	String	map 키 데이터 디코딩을 위한 형식 클래스 스킴을 정의합니다(예: java.lang.Long). key-class 또는 key-type-factory 중 하나를 지정할 수 있습니다. 둘 다 제공되지 않으면 SQL 타입에서 추론하려고 시도합니다(기본 타입만 지원).
fields.#.key-type-factory	optional	(none)	String	map 키 데이터 디코딩을 위한 타입 정보 팩토리를 정의합니다. key-class 또는 key-type-factory 중 하나를 지정할 수 있습니다. 둘 다 제공되지 않으면 SQL 타입에서 추론하려고 시도합니다(기본 타입만 지원).
fields.#.value-class	optional	(none)	String	value 데이터 디코딩을 위한 형식 클래스 스킴을 정의합니다(예: java.lang.Long). value-class 또는 value-info-factory 중 하나를 지정할 수 있습니다. 둘 다 제공되지 않으면 SQL 타입에서 추론하려고 시도합니다(기본 타입만 지원).
fields.#.value-type-factory	optional	(none)	String	value 데이터 디코딩을 위한 타입 정보 팩토리를 정의합니다. value-class 또는 value-type-factory 중 하나를 지정할 수 있습니다. 둘 다 제공되지 않으면 SQL 타입에서 추론하려고 시도합니다(기본 타입만 지원).

기본 데이터 타입 매핑

State SQL 커넥터는 fields.#.value-class와 fields.#.key-class가 정의되지 않았을 때 기본 타입에 대한 데이터 타입을 추론합니다. 다음 표는 Flink SQL type -> Java type의 기본 매핑을 보여줍니다. 매핑이 제대로 계산되지 않으면 열별로 언급된 두 구성 매개변수로 재정의할 수 있습니다.

Flink SQL type	Java type
CHAR / VARCHAR / STRING	java.lang.String
BOOLEAN	boolean
BINARY / VARBINARY	byte[]
DECIMAL	org.apache.flink.table.data.DecimalData
TINYINT	byte
SMALLINT	short
INTEGER	int
BIGINT	long
FLOAT	float
DOUBLE	double
DATE	int
INTERVAL_YEAR_MONTH	long
INTERVAL_DAY_TIME	long
ARRAY	java.util.List
MAP	java.util.Map
ROW	java.util.List<org.apache.flink.table.types.logical.RowType.RowField>

바로가기: STRING SQL 타입의 열에 복잡한 java 클래스가 정의되면 클래스 인스턴스의 toString 메서드 결과가 열 값이 됩니다. 이는 빠른 설명 쿼리가 필요할 때 유용할 수 있습니다.

더 알아보기 (Learn more)

• Operator State
• Savepoint
• Flink SQL 쿼리