Pyspark流媒体

vof42yt1  于 2023-11-16  发布在  Spark
关注(0)|答案(2)|浏览(218)

我在pyspark上写了这个流虹膜分类的代码，但是我得到了这个错误“'RDD'对象没有属性'_jdf'“。我已经将RDD更改为对象框架，但是它告诉我“RDD不是一个itterable”。请帮助我解决它！非常感谢。

这是我的代码：

  1. from pyspark import SparkContext
  2. from pyspark.streaming import StreamingContext
  3. from pyspark.ml import PipelineModel, Pipeline
  4. from pyspark.sql import Row, DataFrame
  5. from pyspark.sql.types import *
  6. from pyspark.sql.functions import *
  8. conf = SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
  9. sc = SparkContext.getOrCreate(conf = conf)
  10. ssc = StreamingContext(sc,1)
  13. lines = ssc.socketTextStream("localhost", 8889)
  15. #Load ML model
  16. sameModel = PipelineModel.load("g:/Demo/DecisionTree_Model1")
  18. #Predict the type of iris from features
  19. result = line.foreachRDD(lambda rdd: sameModel.transform(rdd))
  21. ssc.start()
  22. ssc.awaitTermination()

字符串

和错误：'RDD'对象没有属性'_jdf'

  1. Py4JJavaError                             Traceback (most recent call last)
  2. <ipython-input-6-18f3db416f1c> in <module>()
  3.   1 ssc.start()
  4. ----> 2 ssc.awaitTermination()
  6. E:\Spark\spark\python\pyspark\streaming\context.py in awaitTermination(self, 
  7. timeout)
  8. 204         """
  9. 205         if timeout is None:
  10. --> 206             self._jssc.awaitTermination()
  11. 207         else:
  12. 208             self._jssc.awaitTerminationOrTimeout(int(timeout * 1000))
  14. E:\Spark\spark\python\lib\py4j-0.10.7-src.zip\py4j\java_gateway.py in 
  15. __call__(self, *args)
  16. 1255         answer = self.gateway_client.send_command(command)
  17. 1256         return_value = get_return_value(
  18. -> 1257             answer, self.gateway_client, self.target_id, self.name)
  19. 1258 
  20. 1259         for temp_arg in temp_args:
  22. E:\Spark\spark\python\pyspark\sql\utils.py in deco(*a, **kw)
  23.  61     def deco(*a, **kw):
  24.  62         try:
  25. ---> 63             return f(*a, **kw)
  26.  64         except py4j.protocol.Py4JJavaError as e:
  27.  65             s = e.java_exception.toString()
  29. E:\Spark\spark\python\lib\py4j-0.10.7-src.zip\py4j\protocol.py in 
  30. get_return_value(answer, gateway_client, target_id, name)
  31. 326                 raise Py4JJavaError(
  32. 327                     "An error occurred while calling {0}{1}{2}.\n".
  33. --> 328                     format(target_id, ".", name), value)
  34. 329             else:
  35. 330                 raise Py4JError(
  37. Py4JJavaError: An error occurred while calling o35.awaitTermination.
  38. : org.apache.spark.SparkException: An exception was raised by Python:
  39. Traceback (most recent call last):
  40. File "E:\Spark\spark\python\pyspark\streaming\util.py", line 65, in call
  41. r = self.func(t, *rdds)
  42. File "E:\Spark\spark\python\pyspark\streaming\dstream.py", line 159, in 
  43. <lambda>
  44. func = lambda t, rdd: old_func(rdd)
  45. File "<ipython-input-5-64e27204db5a>", line 1, in <lambda>
  46. result = lines.foreachRDD(lambda rdd: sameModel.transform(rdd))
  47. File "E:\Spark\spark\python\pyspark\ml\base.py", line 173, in transform
  48. return self._transform(dataset)
  49. File "E:\Spark\spark\python\pyspark\ml\pipeline.py", line 262, in _transform
  50. dataset = t.transform(dataset)
  51. File "E:\Spark\spark\python\pyspark\ml\base.py", line 173, in transform
  52. return self._transform(dataset)
  53. File "E:\Spark\spark\python\pyspark\ml\wrapper.py", line 305, in _transform
  54. return DataFrame(self._java_obj.transform(dataset._jdf), dataset.sql_ctx)
  55. AttributeError: 'RDD' object has no attribute '_jdf'
  57. at org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)
  58. at org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)
  59. at org.apache.spark.streaming.api.python.PythonDStream$$anonfun$callForeachRDD$1.apply(PythonDStream.scala:179)
  60. at org.apache.spark.streaming.api.python.PythonDStream$$anonfun$callForeachRDD$1.apply(PythonDStream.scala:179)
  61. at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:51)
  62. at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
  63. at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
  64. at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:416)
  65. at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:50)
  66. at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
  67. at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
  68. at scala.util.Try$.apply(Try.scala:192)
  69. at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)
  70. at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:257)
  71. at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
  72. at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
  73. at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
  74. at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)
  75. at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  76. at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  77. at java.lang.Thread.run(Thread.java:748)

型

pyspark

2条答案

按热度按时间
k2fxgqgv

k2fxgqgv1#

下面的代码展示了如何加载一个预先训练好的模型。用socket的source启动Spark stream，并在其上使用transform。然后将其sink到console。

  1. spark = SparkSession \
  2. .builder \
  3. .appName("transform ml") \
  4. .getOrCreate()
  6. model = PipelineModel.load("./model")
  8. lines = spark \
  9.     .readStream \
  10.     .format("socket") \
  11.     .option("host", "localhost") \
  12.     .option("port", 9999) \
  13.     .load()
  15. random = Random()
  16. words = lines.select(f.lit(random.randint(1, 10000))
  17.                      .alias("id"),
  18.                      lines.value.alias("text")
  19.                      )
  21. prediction = model.transform(words)
  23. query = prediction \
  24.     .writeStream \
  25.     .outputMode("append") \
  26.     .format("console") \
  27.     .start()
  29. query.awaitTermination()

字符串

展开查看全部
赞(0）分享  回复(0）举报  2023-11-16
ubby3x7f

ubby3x7f2#

您看到的错误消息是因为PipelineModel类的transform方法需要DataFrame，而不是RDD。
但是，Spark Streaming中的foreachRDD方法生成的是RDD，而不是DataFrame。要解决这个问题，您需要在应用transform方法之前将RDD转换为DataFrame。以下是您的操作方法：

  1. from pyspark.sql import SparkSession
  3. # Create a SparkSession
  4. spark = SparkSession(sc)
  6. def process(rdd):
  7.     if not rdd.isEmpty():
  8.         df = rdd.toDF()  # Convert RDD to DataFrame
  9.         result = sameModel.transform(df)  # Apply the model
  10.         result.show()  # Display the result
  12. lines.foreachRDD(process)

字符串
在process函数中，我们首先检查RDD是否为空。如果不是，我们将RDD转换为DataFrame，应用模型，然后显示结果。
请注意，toDF方法假设RDD是Row对象或元组列表的形式。如果您的数据不是这种格式，则可能需要在调用toDF之前对其进行转换。
另外，如果没有正确推断，请记住为DataFrame定义模式。您可以通过将StructField对象列表传递给StructType构造函数，然后在调用toDF时使用此模式来完成此操作。

展开查看全部
赞(0）分享  回复(0）举报  2023-11-16
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com