Spark TypeError：LongType 不能接受类型 <type 'unicode'> 中的对象 u'Value'答案

【问题标题】：Spark TypeError: LongType can not accept object u'Value' in type <type 'unicode'>Spark TypeError：LongType 不能接受类型 <type 'unicode'> 中的对象 u'Value'
【发布时间】：2017-02-01 05:00:13
【问题描述】：

我正在使用 spark 将 csv 文件转换为 parquet 格式。我正在使用以下代码。

from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql.types import *


if __name__ == "__main__":
    sc = SparkContext(appName="CSV2Parquet")
    sqlContext = SQLContext(sc)

    schema = StructType([
            StructField("id", LongType(), True),
            StructField("domain", StringType(), True)
           ])

    rdd = sc.textFile("hdfs://10.11.21.33:8020/users2.txt").map(lambda line: line.split(","))
    df = sqlContext.createDataFrame(rdd, schema)
    df.write.parquet('hdfs://10.11.21.33:8020/input-parquet-spark20')

执行因以下错误而停止。

17/02/01 04:54:13 WARN TaskSetManager: Lost task 49.0 in stage 0.0 (TID 49, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/usr/spark/python/lib/pyspark.zip/pyspark/worker.py", line 172, in main
    process()
  File "/usr/spark/python/lib/pyspark.zip/pyspark/worker.py", line 167, in process
    serializer.dump_stream(func(split_index, iterator), outfile)
  File "/usr/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 263, in dump_stream
    vs = list(itertools.islice(iterator, batch))
  File "/usr/spark/python/lib/pyspark.zip/pyspark/sql/session.py", line 510, in prepare
  File "/usr/spark/python/lib/pyspark.zip/pyspark/sql/types.py", line 1350, in _verify_type
    _verify_type(v, f.dataType, f.nullable)
  File "/usr/spark/python/lib/pyspark.zip/pyspark/sql/types.py", line 1322, in _verify_type
    raise TypeError("%s can not accept object %r in type %s" % (dataType, obj, type(obj)))
TypeError: LongType can not accept object u'4168630192959457162' in type <type 'unicode'>

数字 '4168630192959457162' 应该是 64 位整数。这里是 users2.txt 中的示例集

8154738304329264826,"http://0.0.0.0/admin/events/event/
3118660108275961803,"http://127.0.0.1/browser/header/
9223372036854775807,"http://127.0.0.1/account/login
5950027385047304809,"http://127.0.0.1/dashboard/
809124421170478235,"http://127.0.0.1/events/

我是新来的火花。我在这里做错了什么？

【问题讨论】：

标签： python apache-spark spark-dataframe parquet

【解决方案1】：

您需要将字符串编码的数字转换为长：

def parse(line):
    items = line.split(",")
    return (long(items[0]), items[1])

rdd = sc.textFile("hdfs://10.11.21.33:8020/users2.txt").map(parse)

我还会告诉 sc.textFile 不要将 unicode 用于您的用例：

rdd = sc.textFile("hdfs://10.11.21.33:8020/users2.txt", use_unicode=False) #etc

【讨论】：