在 spark 中使用 LSH 对数据帧中的每个点运行最近邻查询
对于数据帧中的每个特征向量，我需要 k 个最近的邻居.我正在使用来自 pyspark 的 BucketedRandomProjectionLSHModel. 创建模型的代码 brp = BucketedRandomProjectionLSH(inputCol="features", outputCol="hashes",seed=12345, bucketLength=n)模型 = brp
..
pyspark-sql相关内容
如何在 PySpark 中的数据帧列中转换 JSON 字符串?
我有一个 pyspark 数据框，如下所示 +--------------------+---+|_c0|_c1|+--------------------+---+|{"object":"F...| 0||{"object":"F...| 1||{"object":"F...| 2||{"object":"E...| 3||{"object":"F...| 4||{"object":"F...
..
py4j.protocol.Py4JJavaError 使用选择语句在数据框中选择嵌套列时出错
我正在尝试在 spark 数据帧 (python) 中执行一个简单的任务，即通过从另一个数据帧中选择特定列和嵌套列来创建新的数据帧例如: df.printSchema()根|-- 时间戳:长(可为空 = 真)|-- 国家:结构体(可为空 = 真)||-- 代码:字符串(可为空 = 真)||-- id: long (nullable = true)||-- time_zone: string (n
..
如何序列化 PySpark GroupedData 对象?
我在具有数百万条记录的数据集上运行 groupBy() 并希望保存结果输出(PySpark GroupedData 对象)，以便我可以- 稍后对其进行序列化并从该点恢复(根据需要在其上运行聚合). df.groupBy("geo_city") 我想避免将 GroupedData 对象转换为 DataFr
..
从 Spark DataFrame 中选择空数组值
给定一个包含以下行的 DataFrame: rows = [行(col1='abc', col2=[8], col3=[18], col4=[16]),行(col2='def', col2=[18], col3=[18], col4=[]),行(col3='ghi', col2=[], col3=[], col4=[])] 对于col2、col3 和col4(即第三行)，我想删除带有空数组的行
..
pyspark 更改日期时间列中的日期
此代码试图更改日期时间列的日期有什么问题 导入pyspark将 pyspark.sql.functions 导入为 sf导入 pyspark.sql.types 作为 sparktypes导入日期时间sc = pyspark.SparkContext(appName="test")sqlcontext = pyspark.SQLContext(sc)rdd = sc.parallelize([(
..
在 Spark 中分组和标准化
我有以下数据框: 将pandas导入为pd将 numpy 导入为 npdf = pd.DataFrame([[1,2,3],[1,2,1],[1,2,2],[2,2,2],[2,3,2],[2,4,2]],columns=["a","b","c"])df = df.set_index("a")df.groupby("a").mean()df.groupby("a").std() 我想标准化
..
PySpark:调用 o51.showString 时出错.没有名为 XXX 的模块
我的 pyspark 版本是 2.2.0.我遇到了一个奇怪的问题.我尝试将其简化如下.文件结构: |root|-- cast_to_float.py|-- 测试|-- 测试.py 在cast_to_float.py中，我的代码: from pyspark.sql.types import FloatType从 pyspark.sql.functions 导入 udfdef cast_to_f
..
在pyspark中左外连接后删除功能不起作用
我的 pyspark 版本是 2.1.1.我正在尝试加入具有两列 id 和 priority 的两个数据帧(左外).我正在像这样创建我的数据框: a = "选择 123 作为 id，1 作为优先级"a_df = spark.sql(a)b = "选择 123 作为 id，1 作为优先级联合选择 112 作为 uid，1 作为优先级"b_df = spark.sql(b)c_df = a_df.j
..
在pyspark中左外连接后删除功能不起作用
我的 pyspark 版本是 2.1.1.我正在尝试加入具有两列 id 和 priority 的两个数据帧(左外).我正在像这样创建我的数据框: a = "选择 123 作为 id，1 作为优先级"a_df = spark.sql(a)b = "选择 123 作为 id，1 作为优先级联合选择 112 作为 uid，1 作为优先级"b_df = spark.sql(b)c_df = a_df.j
..
从火花数据框中选择或删除重复的列
给定一个带有重复列名称(例如 A)的 spark 数据框，我无法修改上游或源，我该如何选择、删除或重命名列之一，以便我可以检索列值? df.select('A') 显示了一个不明确的列错误，filter、drop 和 withColumnRenamed 也是如此.如何选择其中一列? 解决方案 经过数小时的研究，我发现的唯一方法是重命名列集，然后以新集作为标题创建另一个数据框.
..
根据条件组合 Spark 数据框列中的多行
我正在尝试根据条件组合火花数据框中的多行: 这是我拥有的数据框(df): |用户名 |qid |row_no |文字 |---------------------------------||1 |1 |这|||1 |2 |是 ||d |2 |1 ||||1 |3 |文字 ||d |2 |2 |球 | 我希望它看起来像这样 |用户名 |qid |row_no |文字 |---------
..
pyspark 中的 df.show() 问题
我有以下代码: 导入pyspark将熊猫导入为 pd从 pyspark.sql 导入 SQLContext从 pyspark.sql.functions 导入 udf从 pyspark.sql.types 导入 IntegerType, StringTypesc = pyspark.SparkContext()sqlCtx = SQLContext(sc)df_pd = pd.DataFrame
..
在 (Py)Spark 中读取 JDBC 源时出现不受支持的数组错误?
尝试将 postgreSQL DB 转换为 Dataframe .以下是我的代码: from pyspark.sql import SparkSessionspark = SparkSession.builder \.appName("连接数据库") \.getOrCreate()jdbcUrl = "jdbc:postgresql://XXXXXX"连接属性 = {“用户":“"，“密码" :
..
PySpark:调用 o51.showString 时出错.没有名为 XXX 的模块
我的 pyspark 版本是 2.2.0.我遇到了一个奇怪的问题.我尝试将其简化如下.文件结构: |root|-- cast_to_float.py|-- 测试|-- 测试.py 在cast_to_float.py中，我的代码: from pyspark.sql.types import FloatType从 pyspark.sql.functions 导入 udfdef cast_to_f
..
pyspark 更改日期时间列中的日期
此代码试图更改日期时间列的日期有什么问题 导入pyspark将 pyspark.sql.functions 导入为 sf导入 pyspark.sql.types 作为 sparktypes导入日期时间sc = pyspark.SparkContext(appName="test")sqlcontext = pyspark.SQLContext(sc)rdd = sc.parallelize([(
..
Python Spark DataFrame:用 SparseVector 替换 null
在 spark 中，我有以下名为“df"的数据框，其中包含一些空条目: +--------------+------------+--------------------+|身份证|特点1|特点2|+-------+--------------------+--------------------+|185|(5,[0,1,4],[0.1,0...| 空||220|(5,[0,2,3],[0.1
..
如何在pyspark中合并具有条件的两列?
我能够对值进行合并和排序，但无法确定值相等时不合并的条件 df = sqlContext.createDataFrame([("foo", "bar","too","aaa"), ("bar", "bar","aaa","foo")],("k", "K","v","V"))列 = df.columnsk = 0对于范围内的我(len(列)):对于范围内的 j(i + 1, len(columns
..
Spark 请求最大计数
我是 Spark 的初学者，我尝试发出一个请求，允许我检索访问量最大的网页. 我的要求如下 mostPopularWebPageDF = logDF.groupBy("webPage").agg(functions.count("webPage").alias("cntWebPage")).agg(functions.max("cntWebPage")).show() 通过这个请求，我只
..
PySpark sqlContext 读取 Postgres 9.6 NullPointerException
尝试使用 PySpark 从 Postgres 数据库读取表.我已经设置了以下代码并验证了 SparkContext 存在: 导入操作系统os.environ['PYSPARK_SUBMIT_ARGS'] = '--driver-class-path/tmp/jars/postgresql-42.0.0.jar --jars/tmp/jars/postgresql-42.0.0.jar pysp
..