标签: pyspark

我正在使用spark 1.4.0-rc2所以我可以使用python 3和spark.如果我添加export PYSPARK_PYTHON=python3到我的.bashrc文件,我可以使用python 3以交互方式运行spark.但是,如果我想在本地模式下运行一个独立程序,我会收到一个错误:

Exception: Python in worker has different version 3.4 than that in driver 2.7, PySpark cannot run with different minor versions

如何为驱动程序指定python的版本？设置export PYSPARK_DRIVER_PYTHON=python3不起作用.

我有一个数据框,列为String.我想在PySpark中将列类型更改为Double类型.

以下是方式,我做了:

toDoublefunc = UserDefinedFunction(lambda x: x,DoubleType())
changedTypedf = joindf.withColumn("label",toDoublefunc(joindf['show']))

只是想知道,这是通过Logistic回归运行的正确方法,我遇到了一些错误,所以我想知道,这是问题的原因.

我是apache spark的新手,显然我在我的macbook中用自制软件安装了apache-spark:

Last login: Fri Jan  8 12:52:04 on console
user@MacBook-Pro-de-User-2:~$ pyspark
Python 2.7.10 (default, Jul 13 2015, 12:05:58)
[GCC 4.2.1 Compatible Apple LLVM 6.1.0 (clang-602.0.53)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
16/01/08 14:46:44 INFO SparkContext: Running Spark version 1.5.1
16/01/08 14:46:46 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
16/01/08 14:46:47 INFO SecurityManager: Changing view acls to: user
16/01/08 14:46:47 INFO …

我正在使用pyspark(Python 2.7.9/Spark 1.3.1)并且有一个数据帧GroupObject,我需要按降序对其进行过滤和排序.试图通过这段代码实现它.

group_by_dataframe.count().filter("`count` >= 10").sort('count', ascending=False)

但它会引发以下错误.

sort() got an unexpected keyword argument 'ascending'

请为Pandas建议pyspark数据帧替代方案df['col'].unique().

我想列出pyspark数据帧列中的所有唯一值.

不是SQL类型的方式(registertemplate然后SQL查询不同的值).

我也不需要groupby->countDistinct,而是想检查该列中的不同VALUES.

查看新的spark数据帧api,目前还不清楚是否可以修改数据帧列.

我怎么会去改变行的值x列y一个数据帧的？

在pandas这将是df.ix[x,y] = new_value

编辑:合并下面所述的内容,您无法修改现有数据框,因为它是不可变的,但您可以返回具有所需修改的新数据框.

如果您只想根据条件替换列中的值,例如np.where:

from pyspark.sql import functions as F

update_func = (F.when(F.col('update_col') == replace_val, new_value)
                .otherwise(F.col('update_col')))
df = df.withColumn('new_column_name', update_func)

如果要对列执行某些操作并创建添加到数据帧的新列:

import pyspark.sql.functions as F
import pyspark.sql.types as T

def my_func(col):
    do stuff to column here
    return transformed_value

# if we assume that my_func returns a string
my_udf = F.UserDefinedFunction(my_func, T.StringType())

df = df.withColumn('new_column_name', my_udf('update_col'))

如果您希望新列与旧列具有相同的名称,则可以添加其他步骤:

df = df.drop('update_col').withColumnRenamed('new_column_name', 'update_col')

>>> a
DataFrame[id: bigint, julian_date: string, user_id: bigint]
>>> b
DataFrame[id: bigint, quan_created_money: decimal(10,0), quan_created_cnt: bigint]
>>> a.join(b, a.id==b.id, 'outer')
DataFrame[id: bigint, julian_date: string, user_id: bigint, id: bigint, quan_created_money: decimal(10,0), quan_created_cnt: bigint]

有两个id: bigint,我想删除一个.我能怎么做？

正如我在Spark Dataframe中所知,多列的名称可以与下面的数据帧快照中显示的名称相同:

[
Row(a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0}), a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0})),
Row(a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0}), a=125231, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0047, 3: 0.0, 4: 0.0043})),
Row(a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0}), a=145831, f=SparseVector(5, {0: 0.0, 1: 0.2356, 2: 0.0036, 3: 0.0, 4: 0.4132})),
Row(a=107831, f=SparseVector(5, {0: 0.0, 1: …

这个问题的目标是记录:

• 在PySpark中使用JDBC连接读取和写入数据所需的步骤

• JDBC源和已知解决方案可能存在的问题

通过小的更改,这些方法应该与其他支持的语言一起使用,包括Scala和R.

我有一个日期pyspark数据帧,其格式为字符串列MM-dd-yyyy,我试图将其转换为日期列.

我试过了:

df.select(to_date(df.STRING_COLUMN).alias('new_date')).show()

我得到一串空值.有人可以帮忙吗？