Преобразование потоковой структуры данных Spark Structure в Pandas DataFrame - PullRequest
Новая
       23

Преобразование потоковой структуры данных Spark Structure в Pandas DataFrame

1 голос
/ 05 марта 2019

У меня установлено приложение Spark Streaming, которое использует тему Kafka, и мне нужно использовать некоторые API, которые используют Pandas Dataframe, но когда я пытаюсь конвертировать его, я получаю это 

: org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;
kafka
        at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.org$apache$spark$sql$catalyst$analysis$UnsupportedOperationChecker$$throwError(UnsupportedOperationChecker.scala:297)
        at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$$anonfun$checkForBatch$1.apply(UnsupportedOperationChecker.scala:36)
        at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$$anonfun$checkForBatch$1.apply(UnsupportedOperationChecker.scala:34)
        at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:127)
        at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.checkForBatch(UnsupportedOperationChecker.scala:34)
        at org.apache.spark.sql.execution.QueryExecution.assertSupported(QueryExecution.scala:63)
        at org.apache.spark.sql.execution.QueryExecution.withCachedData$lzycompute(QueryExecution.scala:74)
        at org.apache.spark.sql.execution.QueryExecution.withCachedData(QueryExecution.scala:72)
        at org.apache.spark.sql.execution.QueryExecution.optimizedPlan$lzycompute(QueryExecution.scala:78)
        at org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:78)
        at org.apache.spark.sql.execution.QueryExecution.completeString(QueryExecution.scala:219)
        at org.apache.spark.sql.execution.QueryExecution.toString(QueryExecution.scala:202)
        at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:62)
        at org.apache.spark.sql.Dataset.withNewExecutionId(Dataset.scala:2832)
        at org.apache.spark.sql.Dataset.collectToPython(Dataset.scala:2809)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
        at py4j.Gateway.invoke(Gateway.java:282)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.GatewayConnection.run(GatewayConnection.java:238)
        at java.lang.Thread.run(Thread.java:745)

Вот мой код Python 

spark = SparkSession\
    .builder\
    .appName("sparkDf to pandasDf")\
    .getOrCreate()

sparkDf = spark.readStream\
    .format("kafka")\
    .option("kafka.bootstrap.servers", "kafkahost:9092")\
    .option("subscribe", "mytopic")\
    .option("startingOffsets", "earliest")\
    .load()


pandas_df =  sparkDf.toPandas()

query = sparkDf.writeStream\
    .outputMode("append")\
    .format("console")\
    .option("truncate", "false")\
    .trigger(processingTime="5 seconds")\
    .start()\
    .awaitTermination()

Теперь я знаю, что создаю другой экземпляр потокового Dataframe, но где бы я ни пытался использовать start () и awaitTermination (), я получаю ту же ошибку.

Есть идеи?

1 Ответ

2 голосов
/ 05 марта 2019

TL; DR Такая операция просто не может работать.

Теперь я знаю, что создаю еще один экземпляр потокового Dataframe

Ну, проблема в том, что вы действительно этого не делаете. toPandas, вызываемый на DataFrame, создает простые локальные, нераспределенные панды DataFrame, в памяти узла драйвера .

Он не только не имеет ничего общего со Spark, но так как абстракция по своей сути несовместима со структурированным потоковым потоком - Pandas DataFrame представляет фиксированный набор кортежей, в то время как структурированный поток представляет собой бесконечный поток кортежей.

Не совсем ясно, чего вы пытаетесь достичь, и это может быть проблемой XY, но если вам действительно нужно использовать Pandas со структурированным потоковым потоком, вы можете попробовать использовать pandas_udf - SCALAR и Варианты GROUPED_MAP совместимы как минимум с базовыми триггерами, основанными на времени (могут поддерживаться и другие варианты, хотя некоторые комбинации явно не имеют никакого смысла, и я не знаю ни одной официальной матрицы совместимости).

...