A
(тот случай когда у флинка писали сразу java апи, скалу потом подтягивали)
Size: a a a
2021 January 22
A
в спарке наоборот было
A
а автозахват TypeInformation из GenericRecord отработал криво
A
def addSource[T: TypeInformation](function: SourceFunction[T]): DataStream[T] = {
require(function != null, "Function must not be null.")
val cleanFun = scalaClean(function)
val typeInfo = implicitly[TypeInformation[T]]
asScalaStream(javaEnv.addSource(cleanFun, typeInfo))
}A
а явно в параметрах скормить typeInfo нету в скала апи
страдать
страдать
RK
Переписывать на Java? Или как-то можно костыльнуть?
RK
A
val javaStream = env.getJavaEnv.addSource(new FlinkKafkaConsumer[GenericRecord](
kafkaTopic, ConfluentRegistryAvroDeserializationSchema.forGeneric(schema, schemaRegistryURL), properties),
new GenericRecordAvroTypeInfo(schema))
val stream = new DataStream[GenericRecord](javaStream)
A
попробуйте
RK
да, сейчас попробую, отходил пострадать)
RK
@xhumanoid храни тебя господь и жену хорошую
A
разведён
RK
работает
RK
разведён
ну тогда гарем тебе султанский
A
вообще у flink java апи более юзабельна, пускай более многословна =\ нюансы разработки
AE
а ни у кого часом не возникало необходимости бродкастнуть пачку geojson мегов этак на 600 на каждый екзекутор? смотрю что щас по логам что крио делает, и какое время уходит на GC, и как-то нехорошо становится
AE
что прямо исходным текстом раскидывать и парсить внутри мапПартишен, что объектом (а дорожная сеть это нехилый такой граф), всё плохо
K
а ни у кого часом не возникало необходимости бродкастнуть пачку geojson мегов этак на 600 на каждый екзекутор? смотрю что щас по логам что крио делает, и какое время уходит на GC, и как-то нехорошо становится
А почему не на драйвере запарсить?
AE
600 мег это сам JSON, в памяти около 7 гигов