Я пытаюсь вычислить матрицу сходства пользователей по их метаданным.После этого вопроса я нашел вектор данных TF-IDF.После этого я его нормализовал.Результат приведен ниже
data.select('norm').take(10)
это результат
Однако, когда я попытался получить blockMatrix из кадра данных по этому коду
mat = IndexedRowMatrix(
data.select("memberid", "norm").rdd.map(lambda row: IndexedRow(row.memberid, row.norm.toArray()))).toBlockMatrix()
Я получил следующую ошибку
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-123-32e41e450739> in <module>()
1 from pyspark.mllib.linalg.distributed import IndexedRow, IndexedRowMatrix
2 mat = IndexedRowMatrix(
----> 3 data.select("memberid", "norm").rdd.map(lambda row: IndexedRow(row.memberid, row.norm.toArray()))).toBlockMatrix()
~/spark/python/pyspark/mllib/linalg/distributed.py in __init__(self, rows, numRows, numCols)
374 # both be easily serialized. We will convert back to
375 # IndexedRows on the Scala side.
--> 376 java_matrix = callMLlibFunc("createIndexedRowMatrix", rows.toDF(),
377 long(numRows), int(numCols))
378 elif (isinstance(rows, JavaObject)
~/spark/python/pyspark/sql/session.py in toDF(self, schema, sampleRatio)
55 [Row(name=u'Alice', age=1)]
56 """
---> 57 return sparkSession.createDataFrame(self, schema, sampleRatio)
58
59 RDD.toDF = toDF
~/spark/python/pyspark/sql/session.py in createDataFrame(self, data, schema, samplingRatio, verifySchema)
520
521 if isinstance(data, RDD):
--> 522 rdd, schema = self._createFromRDD(data.map(prepare), schema, samplingRatio)
523 else:
524 rdd, schema = self._createFromLocal(map(prepare, data), schema)
~/spark/python/pyspark/sql/session.py in _createFromRDD(self, rdd, schema, samplingRatio)
358 """
359 if schema is None or isinstance(schema, (list, tuple)):
--> 360 struct = self._inferSchema(rdd, samplingRatio)
361 converter = _create_converter(struct)
362 rdd = rdd.map(converter)
~/spark/python/pyspark/sql/session.py in _inferSchema(self, rdd, samplingRatio)
329 :return: :class:`pyspark.sql.types.StructType`
330 """
--> 331 first = rdd.first()
332 if not first:
333 raise ValueError("The first row in RDD is empty, "
~/spark/python/pyspark/rdd.py in first(self)
1326 ValueError: RDD is empty
1327 """
-> 1328 rs = self.take(1)
1329 if rs:
1330 return rs[0]
~/spark/python/pyspark/rdd.py in take(self, num)
1308
1309 p = range(partsScanned, min(partsScanned + numPartsToTry, totalParts))
-> 1310 res = self.context.runJob(self, takeUpToNumLeft, p)
1311
1312 items += res
~/spark/python/pyspark/context.py in runJob(self, rdd, partitionFunc, partitions, allowLocal)
931 # SparkContext#runJob.
932 mappedRDD = rdd.mapPartitions(partitionFunc)
--> 933 port = self._jvm.PythonRDD.runJob(self._jsc.sc(), mappedRDD._jrdd, partitions)
934 return list(_load_from_socket(port, mappedRDD._jrdd_deserializer))
935
~/spark/python/lib/py4j-0.10.3-src.zip/py4j/java_gateway.py in __call__(self, *args)
1131 answer = self.gateway_client.send_command(command)
1132 return_value = get_return_value(
-> 1133 answer, self.gateway_client, self.target_id, self.name)
1134
1135 for temp_arg in temp_args:
~/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
61 def deco(*a, **kw):
62 try:
---> 63 return f(*a, **kw)
64 except py4j.protocol.Py4JJavaError as e:
65 s = e.java_exception.toString()
~/spark/python/lib/py4j-0.10.3-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
317 raise Py4JJavaError(
318 "An error occurred while calling {0}{1}{2}.\n".
--> 319 format(target_id, ".", name), value)
320 else:
321 raise Py4JError(
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.runJob.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 105.0 failed 1 times, most recent failure: Lost task 0.0 in stage 105.0 (TID 1063, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/home/aziz/spark/python/lib/pyspark.zip/pyspark/worker.py", line 172, in main
process()
File "/home/aziz/spark/python/lib/pyspark.zip/pyspark/worker.py", line 167, in process
serializer.dump_stream(func(split_index, iterator), outfile)
File "/home/aziz/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 263, in dump_stream
vs = list(itertools.islice(iterator, batch))
File "/home/aziz/spark/python/pyspark/rdd.py", line 1306, in takeUpToNumLeft
yield next(iterator)
File "<ipython-input-123-32e41e450739>", line 3, in <lambda>
File "/home/aziz/spark/python/lib/pyspark.zip/pyspark/mllib/linalg/distributed.py", line 313, in __init__
self.index = long(index)
ValueError: invalid literal for int() with base 10: '006CCBB6-2304-4A52-8DAD-A88729FCC79F'
at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193)
at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:234)
at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152)
at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
at org.apache.spark.scheduler.Task.run(Task.scala:86)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)