Как использовать модель MLlib в потоковых запросах (сбои с «полем», функции «не существует»)?

Question

Я пытаюсь использовать сохраненную модель Mllib для прогнозирования настроений при передаче данных в реальном времени.

Я испробовал все рекомендации, которые нашел, но все равно получаю ошибки. Текущая ошибка: поле «функции» не существует.

Схема обученных данных:

root
 |-- label: double (nullable = true)
 |-- words: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- features: vector (nullable = true)

lines = spark\
        .readStream\
        .format("kafka")\
        .option("kafka.bootstrap.servers", bootstrapServers)\
        .option("subscribe", topics)\
        .load()\
        .selectExpr("CAST(value AS STRING)")
    #<class 'pyspark.sql.dataframe.DataFrame'>

read_data=lines.selectExpr("CAST(value AS STRING) as text")

model_nb = NaiveBayesModel.load("./myNBmodel")

prediction = model_nb.transform(read_data)

print(prediction.schema)

query1 = prediction.writeStream \
            .outputMode("update") \
            .foreach(process_row) \
            .start()

query1.awaitTermination()

prediction = model_nb.transform(read_data)

: Py4JJavaError: Ошибка при вызове o133.transform. : java.lang.IllegalArgumentException: поле «функции» не существует. Доступные поля: текст

Извлеченным данным не нужны функции, чтобы иметь прогноз, верно?

Answer 1

Выбранные данные не нуждаются в функциях, чтобы иметь прогноз, верно?

Это довольно неправильно.

Необработанные данные должны быть "представлены" ивот почему вы должны использовать Spark MLlib ML Pipelines , поэтому Spark выполняет эту «настройку», а не вы:

ML Pipelines обеспечивают единый наборвысокоуровневые API, созданные на основе DataFrames, которые помогают пользователям создавать и настраивать практические конвейеры машинного обучения.

MLlib стандартизирует API для алгоритмов машинного обучения, чтобы упростить объединение нескольких алгоритмов в один конвейер или рабочий процесс.