데이터 타입

Apache Flink의 Python DataStream API에서 데이터 타입은 DataStream 생태계에서 값의 타입을 설명합니다. 연산의 입력 및 출력 타입을 선언하는 데 사용할 수 있으며 시스템이 요소를 직렬화하는 방법을 알려줍니다.

출처: 문서

본문

Pickle 직렬화

타입이 선언되지 않으면 데이터는 Pickle을 사용해 직렬화 또는 역직렬화됩니다. 예를 들어 다음 프로그램은 데이터 타입을 지정하지 않습니다.

from pyflink.datastream import StreamExecutionEnvironment


def processing():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(1)
    env.from_collection(collection=[(1, 'aaa'), (2, 'bbb')]) \
        .map(lambda record: (record[0]+1, record[1].upper())) \
        .print()  # note: print to stdout on the worker machine

    env.execute()


if __name__ == '__main__':
    processing()

그러나 다음 경우에는 타입을 지정해야 합니다.

Python 레코드를 Java 연산에 전달할 때.
직렬화 및 역직렬화 성능을 개선할 때.

Python 레코드를 Java 연산에 전달

Java 연산자나 함수는 Python 데이터를 식별할 수 없으므로 처리하기 위해 Python 타입을 Java 타입으로 변환하는 데 도움이 되는 타입을 제공해야 합니다. 예를 들어 Java로 구현된 FileSink를 사용해 데이터를 출력하려면 타입을 제공해야 합니다.

from pyflink.common.serialization import Encoder
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.file_system import FileSink


def file_sink():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(1)
    env.from_collection(collection=[(1, 'aaa'), (2, 'bbb')]) \
        .map(lambda record: (record[0]+1, record[1].upper()),
             output_type=Types.ROW([Types.INT(), Types.STRING()])) \
        .add_sink(FileSink
                  .for_row_format('/tmp/output', Encoder.simple_string_encoder())
                  .build())

    env.execute()


if __name__ == '__main__':
    file_sink()

직렬화 및 역직렬화 성능 개선

데이터가 Pickle을 통해 직렬화 및 역직렬화될 수 있더라도 타입이 제공되면 성능이 더 좋아집니다. 명시적 타입을 사용하면 PyFlink가 파이프라인을 통해 레코드를 이동할 때 효율적인 직렬화기를 사용할 수 있습니다.

지원되는 데이터 타입

Python DataStream API에서 타입을 정의하려면 pyflink.common.typeinfo.Types를 사용할 수 있습니다. 아래 표는 현재 지원되는 타입과 정의 방법을 보여줍니다.

PyFlink Type	Python Type	Java Type
`Types.BOOLEAN()`	`bool`	`java.lang.Boolean`
`Types.BYTE()`	`int`	`java.lang.Byte`
`Types.SHORT()`	`int`	`java.lang.Short`
`Types.INT()`	`int`	`java.lang.Integer`
`Types.LONG()`	`int`	`java.lang.Long`
`Types.FLOAT()`	`float`	`java.lang.Float`
`Types.DOUBLE()`	`float`	`java.lang.Double`
`Types.CHAR()`	`str`	`java.lang.Character`
`Types.STRING()`	`str`	`java.lang.String`
`Types.BIG_INT()`	`int`	`java.math.BigInteger`
`Types.BIG_DEC()`	`decimal.Decimal`	`java.math.BigDecimal`
`Types.INSTANT()`	`pyflink.common.time.Instant`	`java.time.Instant`
`Types.TUPLE()`	`tuple`	`org.apache.flink.api.java.tuple.Tuple0` ~ `Tuple25`
`Types.ROW()`	`pyflink.common.Row`	`org.apache.flink.types.Row`
`Types.ROW_NAMED()`	`pyflink.common.Row`	`org.apache.flink.types.Row`
`Types.MAP()`	`dict`	`java.util.Map`
`Types.PICKLED_BYTE_ARRAY()`	The actual unpickled Python object	`byte[]`
`Types.SQL_DATE()`	`datetime.date`	`java.sql.Date`
`Types.SQL_TIME()`	`datetime.time`	`java.sql.Time`
`Types.SQL_TIMESTAMP()`	`datetime.datetime`	`java.sql.Timestamp`
`Types.LIST()`	list of Python object	`java.util.List`

아래 표는 지원되는 배열 타입을 보여줍니다.

PyFlink Array Type	Python Type	Java Type
`Types.PRIMITIVE_ARRAY(Types.BYTE())`	`bytes`	`byte[]`
`Types.PRIMITIVE_ARRAY(Types.BOOLEAN())`	list of bool	`boolean[]`
`Types.PRIMITIVE_ARRAY(Types.SHORT())`	list of int	`short[]`
`Types.PRIMITIVE_ARRAY(Types.INT())`	list of int	`int[]`
`Types.PRIMITIVE_ARRAY(Types.LONG())`	list of int	`long[]`
`Types.PRIMITIVE_ARRAY(Types.FLOAT())`	list of float	`float[]`
`Types.PRIMITIVE_ARRAY(Types.DOUBLE())`	list of float	`double[]`
`Types.PRIMITIVE_ARRAY(Types.CHAR())`	list of str	`char[]`
`Types.BASIC_ARRAY(Types.BYTE())`	list of int	`java.lang.Byte[]`
`Types.BASIC_ARRAY(Types.BOOLEAN())`	list of bool	`java.lang.Boolean[]`
`Types.BASIC_ARRAY(Types.SHORT())`	list of int	`java.lang.Short[]`
`Types.BASIC_ARRAY(Types.INT())`	list of int	`java.lang.Integer[]`
`Types.BASIC_ARRAY(Types.LONG())`	list of int	`java.lang.Long[]`
`Types.BASIC_ARRAY(Types.FLOAT())`	list of float	`java.lang.Float[]`
`Types.BASIC_ARRAY(Types.DOUBLE())`	list of float	`java.lang.Double[]`
`Types.BASIC_ARRAY(Types.CHAR())`	list of str	`java.lang.Character[]`
`Types.BASIC_ARRAY(Types.STRING())`	list of str	`java.lang.String[]`
`Types.OBJECT_ARRAY()`	list of Python object	`Array`

데이터 타입