API-интерфейс Apache Flink Python Datastream переходит в паркет ⇐ Python

[Anonymous]

У меня есть тема Kafka, содержащая сообщения в формате JSON. Используя Flink Python API, я пытаюсь обработать эти сообщения и сохранить их в файлах паркета в GCS.
Вот очищенный фрагмент кода:

Код: Выделить всё

class Extract(MapFunction):
def map(self, value):
record = json.loads(value)
dt_object = datetime.strptime(record['ts'], "%Y-%m-%dT%H:%M:%SZ")
return Row(dt_object, record['event_id'])



events_schema = DataTypes.ROW([
DataTypes.FIELD("ts", DataTypes.TIMESTAMP()),
DataTypes.FIELD("event_id", DataTypes.STRING())
])


# Main job part
kafka_source = KafkaSource.builder() \

.build()

ds: DataStream = env.from_source(kafka_source, WatermarkStrategy.no_watermarks(), "Kafka Source")

mapped_data = ds.map(Extract(), Types.ROW([Types.SQL_TIMESTAMP(), Types.STRING()]))

sink = (FileSink
.for_bulk_format("gs://",
ParquetBulkWriters.for_row_type(row_type=events_schema))
.with_output_file_config(
OutputFileConfig.builder()
.with_part_prefix("bids")
.with_part_suffix(".parquet")
.build())
.build())

mapped_data.sink_to(sink)

Проблема в том, что при попытке запустить это задание я получаю сообщение об ошибке:

Код: Выделить всё

Java.lang.ClassCastException: class java.sql.Timestamp cannot be cast to class java.time.LocalDateTime (java.sql.Timestamp is in module java.sql of loader 'platform'; java.time.LocalDateTime is in module java.base of loader 'bootstrap')

Итак, проблема в том, что Types.SQL_TIMESTAMP() и DataTypes.TIMESTAMP() несовместимы при переводе в соответствующие классы Java. Но я не вижу другого варианта «типизировать» мое преобразование отображения.
Если вместо

Код: Выделить всё

mapped_data = ds.map(Extract(), Types.ROW([Types.SQL_TIMESTAMP(), Types.STRING()]))

Я использую этот вариант

Код: Выделить всё

mapped_data = ds.map(Extract())

тогда я получаю еще одну ошибку:

Код: Выделить всё

java.lang.ClassCastException: class [B cannot be cast to class org.apache.flink.types.Row ([B is in module java.base of loader 'bootstrap'; org.apache.flink.types.Row is in unnamed module of loader 'app')

Мой вопрос: могу ли я сохранить данные, содержащие временные метки, в формате паркета, используя Flink Python API?

Подробнее здесь: https://stackoverflow.com/questions/787 ... to-parquet