Функции из пользовательского модуля, не работающего в PySpark, но они работают при вводе в интерактивном режиме

У меня есть модуль, который я написал, содержащий функции, которые действуют на PySpark DataFrames. Они выполняют преобразование по столбцам в DataFrame, а затем возвращают новый DataFrame. Вот пример кода, укороченный включать только одну из функций:Функции из пользовательского модуля, не работающего в PySpark, но они работают при вводе в интерактивном режиме

from pyspark.sql import functions as F 
from pyspark.sql import types as t 

import pandas as pd 
import numpy as np 

metadta=pd.DataFrame(pd.read_csv("metadata.csv")) # this contains metadata on my dataset 

def str2num(text): 
    if type(text)==None or text=='' or text=='NULL' or text=='null': 
     return 0 
    elif len(text)==1: 
     return ord(text) 
    else: 
     newnum='' 
     for lettr in text: 
      newnum=newnum+str(ord(lettr)) 
     return int(newnum) 

str2numUDF = F.udf(lambda s: str2num(s), t.IntegerType()) 

def letConvNum(df): # df is a PySpark DataFrame 
    #Get a list of columns that I want to transform, using the metadata Pandas DataFrame 
    chng_cols=metadta[(metadta.comments=='letter conversion to num')].col_name.tolist() 
    for curcol in chng_cols: 
     df=df.withColumn(curcol, str2numUDF(df[curcol])) 
    return df

Так что мой модуль, называют его mymodule.py. Если я начинаю PySpark оболочку, и я делаю следующее:

import mymodule as mm 
myf=sqlContext.sql("select * from tablename lim 10")

проверить Myf (PySpark DataFrame) и это нормально. Я проверяю, что я на самом деле импортирован MyModule, пытаясь использовать функцию str2num:

mm.str2num('a') 
97

Так что на самом деле импортирует модуль. Тогда, если я пытаюсь это:

df2=mm.letConvNum(df)

И сделать это, чтобы проверить, что он работал:

df2.show()

Он пытается выполнить действие, но тогда он выходит из строя:

16/03/10 16:10:44 ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 365) 
    org.apache.spark.api.python.PythonException: Traceback (most recent call last): 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/worker.py", line 98, in main 
     command = pickleSer._read_with_length(infile) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length 
     return self.loads(obj) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads 
     return pickle.loads(obj) 
     File "test2.py", line 16, in <module> 
     str2numUDF=F.udf(lambda s: str2num(s), t.IntegerType()) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1460, in udf 
     return UserDefinedFunction(f, returnType) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1422, in __init__ 
     self._judf = self._create_judf(name) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1430, in _create_judf 
     pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command, self) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2317, in _prepare_for_python_RDD 
     [x._jbroadcast for x in sc._pickled_broadcast_vars], 
    AttributeError: 'NoneType' object has no attribute '_pickled_broadcast_vars' 

      at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166) 
      at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207) 
      at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125) 
      at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute$1.apply(python.scala:397) 
      at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute$1.apply(python.scala:362) 
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$17.apply(RDD.scala:710) 
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$17.apply(RDD.scala:710) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) 
      at org.apache.spark.scheduler.Task.run(Task.scala:88) 
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) 
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
      at java.lang.Thread.run(Thread.java:745) 
    16/03/10 16:10:44 ERROR TaskSetManager: Task 0 in stage 1.0 failed 1 times; aborting job 
    Traceback (most recent call last): 
     File "<stdin>", line 1, in <module> 
     File "/usr/hdp/2.3.4.0-3485/spark/python/pyspark/sql/dataframe.py", line 256, in show 
     print(self._jdf.showString(n, truncate)) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__ 
     File "/usr/hdp/2.3.4.0-3485/spark/python/pyspark/sql/utils.py", line 36, in deco 
     return f(*a, **kw) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value 
    py4j.protocol.Py4JJavaError: An error occurred while calling o7299.showString. 
    : org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1.0 (TID 365, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last): 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/worker.py", line 98, in main 
     command = pickleSer._read_with_length(infile) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length 
     return self.loads(obj) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads 
     return pickle.loads(obj) 
     File "test2.py", line 16, in <module> 
     str2numUDF=F.udf(lambda s: str2num(s), t.IntegerType()) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1460, in udf 
     return UserDefinedFunction(f, returnType) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1422, in __init__ 
     self._judf = self._create_judf(name) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1430, in _create_judf 
     pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command, self) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2317, in _prepare_for_python_RDD 
     [x._jbroadcast for x in sc._pickled_broadcast_vars], 
    AttributeError: 'NoneType' object has no attribute '_pickled_broadcast_vars' 

      at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166) 
      at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207) 
      at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125) 
      at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute$1.apply(python.scala:397) 
      at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute$1.apply(python.scala:362) 
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$17.apply(RDD.scala:710) 
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$17.apply(RDD.scala:710) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) 
      at org.apache.spark.scheduler.Task.run(Task.scala:88) 
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) 
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
      at java.lang.Thread.run(Thread.java:745) 

    Driver stacktrace: 
      at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283) 
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1271) 
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1270) 
      at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 
      at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) 
      at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1270) 
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697) 
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697) 
      at scala.Option.foreach(Option.scala:236) 
      at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:697) 
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1496) 
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1458) 
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447) 
      at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48) 
      at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:567) 
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:1824) 
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:1837) 
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:1850) 
      at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:215) 
      at org.apache.spark.sql.execution.Limit.executeCollect(basicOperators.scala:207) 
      at org.apache.spark.sql.DataFrame$$anonfun$collect$1.apply(DataFrame.scala:1385) 
      at org.apache.spark.sql.DataFrame$$anonfun$collect$1.apply(DataFrame.scala:1385) 
      at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:56) 
      at org.apache.spark.sql.DataFrame.withNewExecutionId(DataFrame.scala:1903) 
      at org.apache.spark.sql.DataFrame.collect(DataFrame.scala:1384) 
      at org.apache.spark.sql.DataFrame.head(DataFrame.scala:1314) 
      at org.apache.spark.sql.DataFrame.take(DataFrame.scala:1377) 
      at org.apache.spark.sql.DataFrame.showString(DataFrame.scala:178) 
      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:497) 
      at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231) 
      at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379) 
      at py4j.Gateway.invoke(Gateway.java:259) 
      at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133) 
      at py4j.commands.CallCommand.execute(CallCommand.java:79) 
      at py4j.GatewayConnection.run(GatewayConnection.java:207) 
      at java.lang.Thread.run(Thread.java:745) 
    Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last): 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/worker.py", line 98, in main 
     command = pickleSer._read_with_length(infile) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length 
     return self.loads(obj) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads 
     return pickle.loads(obj) 
     File "test2.py", line 16, in <module> 
     str2numUDF=F.udf(lambda s: str2num(s), t.IntegerType()) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1460, in udf 
     return UserDefinedFunction(f, returnType) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1422, in __init__ 
     self._judf = self._create_judf(name) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1430, in _create_judf 
     pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command, self) 
     File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2317, in _prepare_for_python_RDD 
     [x._jbroadcast for x in sc._pickled_broadcast_vars], 
    AttributeError: 'NoneType' object has no attribute '_pickled_broadcast_vars' 

      at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166) 
      at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207) 
      at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125) 
      at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute$1.apply(python.scala:397) 
      at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute$1.apply(python.scala:362) 
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$17.apply(RDD.scala:710) 
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$17.apply(RDD.scala:710) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) 
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) 
      at org.apache.spark.scheduler.Task.run(Task.scala:88) 
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) 
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
      ... 1 more

Как check, я открыл чистую оболочку и вместо импорта модуля я просто определил функцию str2num и UDF в интерактивной оболочке. Затем я напечатал в содержании последней функции, и сделал то же окончательную проверку:

df2.show()

На этот раз, я получаю обратно преобразованный DataFrame я ожидал.

Почему это работает, когда функции вводятся интерактивно, но не когда они считываются из модуля? Я знаю, что он читает модуль, так как работает регулярная функция str2num.

источник

2016-03-10 RKD314

И что именно находится в _bunch of errors_? – zero323

Я редактировал вопрос, чтобы включить ошибки. – RKD314

У меня была такая же ошибка и последовала за трассировкой стека.

В моем случае я создавал файл с яйцом, а затем передавал его искру через опцию --py-files.

Что касается ошибки, я думаю, что это сводится к тому, что при вызове F.udf(str2num, t.IntegerType()) экземпляр UserDefinedFunction создаются перед тем Спарк работает, поэтому она имеет пустую ссылку на какое-то SparkContext, назовет его sc. Когда вы запускаете UDF, ссылается sc._pickled_broadcast_vars, и это выдает AttributeError в вашем выводе.

Моя работа заключается в том, чтобы избежать создания UDF до запуска Spark (и, следовательно, есть активный SparkContext.В вашем случае, вы можете просто изменить свое определение

def letConvNum(df): # df is a PySpark DataFrame 
    #Get a list of columns that I want to transform, using the metadata Pandas DataFrame 
    chng_cols=metadta[(metadta.comments=='letter conversion to num')].col_name.tolist() 

    str2numUDF = F.udf(str2num, t.IntegerType()) # create UDF on demand 
    for curcol in chng_cols: 
     df=df.withColumn(curcol, str2numUDF(df[curcol])) 
    return df

Примечание: На самом деле я не проверял код, приведенный выше, но изменение в моем собственном коде был похож, и все работало нормально.

Кроме того, для заинтересованного читателя, увидеть Spark code for UserDefinedFunction

источник

2016-10-05 15:50:41

Какая версия искры вы на btw?

Преобразование функции в UDF как это:

str2numUDF = F.udf(str2num, t.IntegerType())

Вам не нужна функция лямбда здесь.

источник

2016-08-11 07:07:10 Max

Да, это правда. Это не решает проблему, хотя (я пробовал решение, которое вы предлагаете). Я больше не пытаюсь это решить, но я был на Spark 1.6. – RKD314

Если вы только когда-либо использовать UDF внутри других функций, вы могли бы сделать, как это.

from pyspark.sql.functions import udf 


class Udf(object): 
    def __init__(s, func, spark_type): 
     s.func, s.spark_type = func, spark_type 

    def __call__(s, *args): 
     return udf(s.func, s.spark_type)(*args) 


myfunc_udf = Udf(myfunc, StringType()) 


def processing(): 
    df_new = df.select(myfunc_udf('somefield'))

источник

2016-10-26 16:01:01 innohead

Вместо того, чтобы создавать класс, я закончил писать лямбда для каждого из моих UDF - myfunc_udf = lambda arg: udf (f = myfunc, returnType = StringType()) (arg) –

Я пробиваю себе голову по этой проблеме в течение 20 часов. Спасибо за ребята!

вот мой вариант, в случае, если кто-то заинтересован в том, как я решил ту же проблему. хотя в основном он получен из кода/ответов выше.

Цель состоит в том, чтобы просто преобразовать столбцы строки, чтобы показать их длину, но вы можете делать все, что угодно (я проверяю данные и отслеживание ошибок в моем основном приложении).

то, что я использую udf for, намного сложнее, однако это то, что я сделал, чтобы проверить, что моя udf теперь работает.

Если предположить, что dataframe это все StringType() в моем случае я имел 4 столбцы со строками

Решение:

Я сделал отдельный .py файл с именем myfunctions

внутри него

from pyspark.sql import functions as F 
from pyspark.sql.types import IntegerType 
import logging 

def str2num(text): 
    if type(text) == None or text == '' or text == 'NULL' or text == 'null': 
     return 0 
    else: 
     return len(text) 


def letConvNum(df, columns): 
    str2numUDF = F.udf(str2num, IntegerType()) 
    logging.info(columns) 
    index = 0 
    for curcol in columns: 
     df = df.withColumn(curcol, str2numUDF(df[curcol])) 
     index += 1 
    return df

Затем внутри моего основного класса добавьте новый .py-файл в свой sparkContext

#my understanding is that this insures your function is added to a spark across all nodes 
sc.addPyFile("./myfunctions.py") 

#dynamically create headers based on config -simplified for example 
schemaString = "YearMonth,IMEI,IMSI,MSISDN" 
fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split(",")] 
schema = StructType(fields) 

df = sqlContext.read.format('com.databricks.spark.csv').options(header='false', inferschema='false', delimiter='|').load('/app/teacosy/invictus/kenya/SAF_QUALCOMM_IMEI_20170321.txt', schema=schema) 

#read and write file to get parquet. please note this was to optimize MASSIVE files 50-200g 
df.write.parquet("data.parquet", mode='overwrite') 
dataframe = sqlContext.read.parquet("data.parquet") 

df2 = mf.letConvNum(dataframe, schemaString.split(",")) 
df2.show()

Вход:

+---------+---------------+---------------+------------+ 
|YearMonth|   IMEI|   IMSI|  MSISDN| 
+---------+---------------+---------------+------------+ 
| 201609|869859025975610|639021005869699|254724884336| 
| 201609|359521062182040|639021025339132|254721224577| 
| 201609|353121070662770|639021025339132|254721224577| 
| 201609|868096015837410|639021025339132|254721224577| 
| 201609|866204020015610|639021025339132|254721224577| 
| 201609|356051060479107|639028040455896|254710404131| 
| 201609|353071062803703|639027641207269|254725555262| 
| 201609|356899067316490|639027841002602|254711955201| 
| 201609|860357020164930|639028550063234|254715570856| 
| 201609|862245026673900|639028940332785|254728412070| 
| 201609|352441075290910|639029340152407|254714582871| 
| 201609|862074027499277|639029340152407|254714582871| 
| 201609|357036073532528|639028500408346|254715408346| 
| 201609|356546060475230|639021011628783|254722841516| 
| 201609|356546060475220|639021011628783|254722841516| 
| 201609|866838023727117|639028840277749|254718492024| 
| 201609|354210053950950|639029440054836|254729308302| 
| 201609|866912020393040|639029870328080|254725528182| 
| 201609|357921070054540|639028340694869|254710255083| 
| 201609|357977056264767|639027141561199|254721977494|

Выход:

+---------+----+----+------+ 
|YearMonth|IMEI|IMSI|MSISDN| 
+---------+----+----+------+ 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12| 
|  6| 15| 15| 12|

Я надеюсь, что это помогает любому борется с видя их pyspark приложений заморозить или повесить ... так расстраивает омг ...

источник

2018-02-01 07:25:47 Elton

Функции из пользовательского модуля, не работающего в PySpark, но они работают при вводе в интерактивном режиме

ответ

Смежные вопросы