Spark dataframe:使用第二个dataframe查找数组元素
Spark DataFrame是一种分布式数据集，可以以表格形式组织和处理数据。它提供了丰富的API和函数，用于数据的转换、过滤、聚合和分析。
在Spark DataFrame中，使用第二个DataFrame查找数组元素可以通过以下步骤实现：
- 导入必要的库和模块：from pyspark.sql import SparkSession from pyspark.sql.functions import col, array_contains
- 创建SparkSession对象：spark = SparkSession.builder.appName("Array Element Lookup").getOrCreate()
- 创建第一个DataFrame：data = [("John", [1, 2, 3]), ("Alice", [4, 5, 6]), ("Bob", [7, 8, 9])] df1 = spark.createDataFrame(data, ["name", "numbers"]) df1.show()这将创建一个包含两列（name和numbers）的DataFrame，并显示其内容。
- 创建第二个DataFrame：lookup_data = [("John", 2), ("Alice", 5), ("Bob", 10)] df2 = spark.createDataFrame(lookup_data, ["name", "lookup_number"]) df2.show()这将创建一个包含两列（name和lookup_number）的DataFrame，并显示其内容。
- 使用第二个DataFrame查找数组元素：result = df1.join(df2, on=["name"]).filter(array_contains(col("numbers"), col("lookup_number"))) result.show()这将使用join操作将两个DataFrame连接起来，并使用filter函数查找包含指定元素的数组。最后，显示结果。
Spark DataFrame的优势在于其分布式计算能力和丰富的API，可以处理大规模数据集并进行复杂的数据操作和分析。它适用于各种场景，包括数据清洗、数据挖掘、机器学习和实时数据处理等。
腾讯云提供了一系列与Spark相关的产品和服务，例如TencentDB for Apache Spark、Tencent Cloud Data Lake Analytics等。您可以通过访问腾讯云官方网站（https://cloud.tencent.com/）了解更多相关产品和详细信息。
相关·内容
为了在大型数据集上获得一些离群点图，我需要将spark DataFrame转换为pandas。Turing to Apache Arrow在将x转换为字符串时，一次简单的运行会使我的pyspark控制台崩溃(它在没有转换的情况下工作得很好)，为什么？
Using Python version 3.8.9 (default, Apr 10 2021 15:47:22)
Spark context Web UI available at http://6d0b1018a45a:4040
Spark context available as 'sc' (master = local[
浏览 20提问于2021-05-16得票数 1
回答已采纳
1回答
我试图在星火应用程序中映射来自mongodb连接器的数据。在此之前，我没有其他错误，所以我假设到mongodb的连接是成功的。我使用以下代码来映射：
JavaRDD<AppLog> logs = documents.map(
new Function<Tuple2<Object, BSONObject>, AppLog>() {
public AppLog call(final Tuple2<Object, BSONObject> tuple) {
AppLog log = new AppLog();
浏览 5提问于2016-10-13得票数 1
回答已采纳
2回答
使用PySpark 1.6.3，我试图将RDD转换为Dataframe。这是在齐柏林飞艇笔记本上运行的测试代码。感兴趣的是rdd_ret。
>>> from pyspark.sql import Row
>>> rdd_ret.count()
9301
>>> rddofrows = rdd_ret.map(lambda x: Row(**x))
>>> things = rddofrows.take(10000)
>>> len(things)
9301
>>> [type(x) fo
浏览 0提问于2018-06-21得票数 2
假设我们有一个具有以下模式的PySpark dataframe df
root
|-- a1: string (nullable = true)
|-- a2: string (nullable = true)
|-- arr1: array (nullable = true)
| |-- element: struct (containsNull = true)
| | |-- dt1: date (nullable = true)
| | |-- dt2: date (nullable = true)
| | |-- dt_indic
浏览 12提问于2021-12-11得票数 0
1回答
我编写了将字符串转换为结构数组的代码。我也想在python上做同样的事情。你知道我该怎么做吗？
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._
import org.apache.spark.sql.Column
val df: DataFrame = Seq(
"adserviceCalculateCpcAlgorithmV1:2;searchProductsDecorator:3;searchOffersDecorator:3;bundlediscounts:5;se
浏览 3提问于2022-08-02得票数 0
回答已采纳
4回答
我有一个只有一列的Dataframe该列的每一行都有一个字符串值数组：
Spark2.2 Dataframe中的值
["123", "abc", "2017", "ABC"]
["456", "def", "2001", "ABC"]
["789", "ghi", "2017", "DEF"]
org.apache.spark.sql.DataFrame = [col: array]
root
|--
浏览 8提问于2017-12-01得票数 28
回答已采纳
我试图在Dataframe中转换json文件，但我被困在基于数组的字段中。我使用的是Spark 1.6和Java。当我读取嵌套的Json并转换为Dataframe时，我可以读取一些字段，但当我尝试输入特定路径时，会显示错误。 DataFrame df = spark.read().json(sc.wholeTextFiles("PATH").values()); 我正在读取一个json文件。 df.select(col("orcamentos.itens")).printSchema();
root
|-- itens: array (nullable =
浏览 29提问于2019-09-11得票数 0
我从spark数组“df_spark”开始： from pyspark.sql import SparkSession
import pandas as pd
import numpy as np
import pyspark.sql.functions as F
spark = SparkSession.builder.master("local").appName("Word Count").config("spark.some.config.option", "some-value").getOrCreate()
np
浏览 19提问于2018-12-16得票数 3
回答已采纳
1回答
我在Spark2.2.0和Scala2.11.8中有下面的DataFrame。
+----------+-------------------------------+
|item | other_items |
+----------+-------------------------------+
| 111 |[[444,1.0],[333,0.5],[666,0.4]]|
| 222 |[[444,1.0],[333,0.5]] |
| 333 |[]
浏览 1提问于2017-11-22得票数 4
回答已采纳
我有一个带有两个列的Spark (Python) dataframe :一个用户ID，然后一个数组数组，它在Spark中表示为一个包装的数组，如下所示：
[WrappedArray(9, 10, 11, 12), WrappedArray(20, 21, 22, 23, 24, 25, 26)]
在其通常表示形式中，如下所示：
[[9, 10, 11, 12], [20, 21, 22, 23, 24, 25, 26]]
我想对每个子数组执行操作，例如，取第三个列表并检查它的任何值是否在第一个子数组中，但我似乎找不到针对pyflem2.0的解决方案(只有Scala特有的老解决方案，如和)。
浏览 2提问于2017-04-06得票数 1
回答已采纳
2回答
我目前是.NET的新手，需要将C#列表追加到增量表中。我假设我首先需要创建一个Spark DataFrame来做这件事。在示例代码中，如何将“name”附加到dataframe "df"？
现在这似乎已经被弃用了()，新版本()中不能使用RDD
using System.Collections.Generic;
using Microsoft.Spark.Sql;
namespace HelloSpark
{
class Program
{
static void Main(string[] args)
{
浏览 22提问于2019-08-08得票数 1
我正在使用Databricks 4.3 (包括ApacheSpark2.3.1，Scala2.11)，Python版本3.5。
我有一个火花数据框架df_spark，我运行熊猫分组UDF，以获得一个新的火花数据框架df_spark2，其中只有一列的字符串类型。当我显示df_spark2的头部时，我得到了错误：
org.apache.spark.SparkException: Job aborted due to stage failure: Task 18 in stage 12.0 failed 4 times, most recent failure: Lost task 18.3 in
浏览 1提问于2019-01-23得票数 1
有时(例如，为了测试和标记)，我想强制执行在DataFrame上定义的转换。AFAIK调用像count这样的操作并不能确保所有的Columns都是实际计算的，show可能只计算所有Rows的一个子集(参见下面的示例) 我的解决方案是使用df.write.saveAsTable将DataFrame写到HDFS，但是这会“扰乱”我的系统，我不想再保存更多的表。 那么触发DataFrame求值的最佳方式是什么呢 编辑： 请注意，在spark开发人员列表上还有一个最近的讨论：http://apache-spark-developers-list.1001551.n3.nabble.com/Will-
浏览 62提问于2017-03-10得票数 20
回答已采纳
我已经从np.select获得了一个numpy数组，我希望将其存储为PySpark DataFrame中的一个新列。我怎么能这么做？
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
pdf = pd.DataFrame({'a': [1,2,3], 'b': ['abc', 'cde', 'edf']})
df_data = spark.createDataFrame(pdf, schema='
浏览 6提问于2022-05-25得票数 1
我使用以下命令创建了一个空的Seq() scala> var x = Seq[DataFrame]()
x: Seq[org.apache.spark.sql.DataFrame] = List() 我有一个名为createSamplesForOneDay()的函数，它返回一个DataFrame，我想将它添加到这个Seq() x中。 val temp = createSamplesForOneDay(some_inputs) // this returns a Spark DF
x = x + temp // this throws an error 我得到下面的错误- scala&
浏览 57提问于2019-07-02得票数 0
回答已采纳
我在Spark中有以下DataFrame (我正在使用Scala)：
[[1003014, 0.95266926], [15, 0.9484202], [754, 0.94236785], [1029530, 0.880922], [3066, 0.7085166], [1066440, 0.69400793], [1045811, 0.663178], [1020059, 0.6274495], [1233982, 0.6112905], [1007801, 0.60937023], [1239278, 0.60044676], [1000088, 0.5789191], [1056268,
浏览 1提问于2019-12-09得票数 1
回答已采纳
1回答
我有一个DataFrame (转换为RDD)，并希望重新分区，以便每个键(第一列)都有自己的分区。这是我所做的：
# Repartition to # key partitions and map each row to a partition given their key rank
my_rdd = df.rdd.partitionBy(len(keys), lambda row: int(row[0]))
但是，当我试图将它映射回DataFrame或保存它时，我得到了这个错误：
Caused by: org.apache.spark.api.python.PythonException:
浏览 0提问于2015-11-21得票数 5
3回答
火花:与阵列连接
、、
我需要将一个带有string列的dataframe连接到一个字符串数组中，这样如果数组中的值匹配，行就会连接起来。
我试过了，但我想这不是支持。还有别的办法吗？
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("test")
val spark = SparkSession.builder().config(sparkConf).g
浏览 1提问于2017-08-07得票数 10
回答已采纳
在这里，我试图将时间戳动态地添加到数据帧中，
{“行动”：“事件”、"id":1173、"lat":0.0、"lon":0.0、"rollid":55、"event":"type“、"ccd":0、"fone":"ione”、"version":"10.1“、"item":"shack"}
在上面输入的数据中，im试图用下面的代码追加时间戳
foreachRDD(rdd=>
74
浏览 1提问于2017-01-17得票数 0
下面是spark shell代码
scala> val colName = "time_period_id"
scala> val df = spark.sql("""select time_period_id from prod.demo where time_period_id =
202101102 """)
df: org.apache.spark.sql.DataFrame = [time_period_id: int]
scala> val result = df.agg(max(
浏览 24提问于2021-03-11得票数 2
回答已采纳
1回答
因为我是新手，所以我有一个简单的疑问
我必须创建一个空的dataframe，稍后我必须根据一些条件填充它。
我已经经历了创建空数据帧的许多问题，但是下面这些方法之间的区别是什么
我不知道这是不是正确的方法
def function1(df: DataFrame): DataFrame = {
var newdf:DataFrame= null;
if(!x._2(0).column.trim.isEmpty){
newdf= spark.sql("SELECT f_name,l_name FROM tab1");
}else{
浏览 1提问于2020-01-20得票数 1
我已经创建了一个空的dataframe，并开始添加它，通过读取每个文件。但其中一个文件的列数比前一个文件多。如何仅为所有其他文件选择第一个文件中的列？
from pyspark.sql import SparkSession
from pyspark.sql import SQLContext
from pyspark.sql.types import StructType
import os, glob
spark = SparkSession.builder.\
config("spark.jars.packages","saurf
浏览 1提问于2019-09-06得票数 3
回答已采纳
1回答
我的目标是在火花中做一些文本挖掘。我使用的是SPark 2.1。为此，我需要读取文本文件，并将它们保存为RDD/DataFrame的元素。我的问题是，在spark中，每一行都被解释为一个元素，但我希望每个文本文件都获得RDD的一个元素。我所写的代码：
from pyspark.sql import SparkSession
from pyspark.ml.feature import StringIndexer
from pyspark.ml import Pipeline
spark = SparkSession\
.builder\
.getOrCreate()
浏览 2提问于2017-08-08得票数 1
我正在从“雪花”中的存储表中读取数据。我希望将日期从dataframe maxdatefromtbl传递给我在spark中的查询，以筛选记录。此条件(s"CREATED_DATE!='$maxdatefromtbl'")未按预期工作。
var retail = spark.read.format("snowflake").options(options).option("query","Select MAX(CREATED_DATE) as CREATED_DATE from RSTORE").load()
va
浏览 4提问于2022-10-26得票数 0
回答已采纳
我有以下示例数据
fruit_list = ['apple', 'apple', 'orange', 'apple']
qty_list = [16, 2, 3, 1]
spark_df = spark.createDataFrame([(101, 'Mark', fruit_list, qty_list)], ['ID', 'name', 'fruit', 'qty'])
我想创建另一个专栏，其中包含一个类似于熊猫groupby('fruit
浏览 0提问于2019-07-31得票数 3
回答已采纳
我使用的是spark 1.6。我需要为dataframe中的一列找到多个百分位数。我的数据非常庞大，至少有1000万条记录。我尝试使用配置单元上下文，如下所示
hivecontext.sql("select percentile_approx(col,0.25),percentile_approx(col,0.5) from table")
但是这种方法非常慢，而且需要大量的时间。我听说过approxQuantile，但似乎它在spark 2.x中可用。在spark 1.6中，是否有使用spark dataframe的替代方法来提高性能。
我看到了另一种使用hive UDAF的
浏览 14提问于2019-10-31得票数 0
回答已采纳
1回答
Spark 2.1版
我正在将一个文件读入Spark dataframe，格式如下：
{
"field1": "value1",
"field2": "value2",
"elements": [{
"id": "1",
"name": "a"
},
{
"id": "2",
"name": "
浏览 11提问于2017-02-22得票数 1
回答已采纳
1回答
我在Spark中有一个dataframe (通过读取csv创建)，如何在C#中遍历这个数据文件中的行。dataframe中有10行3列，在逐行导航时，我希望获得每个列的值。以下是我正在尝试的：
foreach (var obj in df)
{
Console.WriteLine("test");
}
foreach语句不能对“DataFrame”类型的变量进行操作，因为“DataFrame”不包含“GetEnumerator”的公共实例定义
浏览 4提问于2022-11-09得票数 0
回答已采纳
1回答
这是推荐的，以及为什么性能方面的spark.dataframe.count()或spark.dataframe.take(1)。
浏览 1提问于2019-02-06得票数 0
回答已采纳
我有一个简单的微软( dotnet )星火应用程序，它在使用UDFs.I时抛出System.NullReferenceException，我刚刚开始学习使用dotnet的火花。这个应用程序看起来很简单，但我不知道我在这里错过了什么。
以下是主要的方法：
static void Main(string[] args)
{
SparkSession spark = SparkSession
.Builder()
.AppName("Sample app")
浏览 14提问于2022-03-19得票数 0
我需要检索每个dataframe分区的第一个元素。我知道我需要使用mapPartitions，但我不清楚如何使用它。
注意:我使用的是Spark2.0，对数据进行排序。
浏览 3提问于2016-09-28得票数 0
回答已采纳
除了下面使用Scala的示例之外，我没有找到任何用于将矩阵转换为spark dataframe的pyspark代码。有人知道如何使用python吗？ How to convert a mllib matrix to a spark dataframe?
浏览 61提问于2019-01-09得票数 1
回答已采纳
2回答
我是Scala编程的新手，这是我的问题:如何计算每行的字符串数？我的数据帧由ArrayString类型的单个列组成。
friendsDF: org.apache.spark.sql.DataFrame = [friends: array<string>]
浏览 2提问于2017-09-07得票数 45
回答已采纳
1回答
如何在没有方括号的情况下获得值
、、、、
我已经使用Scala和Spark SQL创建了一个数据帧。我想要表中的第一个值，但我将它放在方括号[]中。我能不能只得到没有括号的值？
代码：
val sigh = sqlContext.sql("""SELECT DISTINCT accountId FROM table""")
sigh.show()
sigh.select("accountId").first()
结果：
sigh: org.apache.spark.sql.DataFrame = [accountId: string]
+----------+
| acc
浏览 21提问于2018-08-22得票数 0
回答已采纳
1回答
我想将dataframe的架构拆分为一个集合。我正在尝试这个方法，但是模式被打印成一个字符串。我是否可以将其拆分为每个StructType的集合，以便对其进行操作(比如只从输出中提取数组列)？我试图平平一个复杂的多层次的结构+数组数据。
import org.apache.spark.sql.functions.explode
import org.apache.spark.sql._
val test = sqlContext.read.json(sc.parallelize(Seq("""{"a":1,"b":[2,3],"
浏览 0提问于2018-03-09得票数 0
回答已采纳
3回答
给定具有不同大小的整数数组列的数据文件：
scala> sampleDf.show()
+------------+
| arrays|
+------------+
|[15, 16, 17]|
|[15, 16, 17]|
| [14]|
| [11]|
| [11]|
+------------+
scala> sampleDf.printSchema()
root
|-- arrays: array (nullable = true)
| |-- element: integer (containsNull =
浏览 0提问于2018-05-22得票数 3
回答已采纳
我想像这样把预测结果写回oracle数据库: model.transform(testDate).write.mode(SaveMode.Overwrite).jdbc(url，"b_spark_tst"，DataFrame )
但是我得到了这个错误信息：
Exception in thread "main" java.lang.IllegalArgumentException: Can't get JDBC type for array<string>
谁能帮我把DataFrames写到数据库里？
谢谢!
更新：
这是我的DataFrame
浏览 3提问于2016-11-30得票数 2
我是PySpark的新手。
我有一个JSON文件，它在架构下面
df = spark.read.json(input_file)
df.printSchema()
|-- UrlsInfo: array (nullable = true)
| |-- element: struct (containsNull = true)
| | |-- displayUrl: string (nullable = true)
| | |-- type: string (nullable = true)
| | |-- url: string (nul
浏览 0提问于2019-03-13得票数 0
回答已采纳
1回答
我正在使用一个Dataframe，它有一个类似于此的复杂模式：
root
|-- NPAData: struct (nullable = true)
| |-- NPADetails: struct (nullable = true)
| | |-- location: string (nullable = true)
| | |-- manager: string (nullable = true)
| |-- usersDetails: array (nullable = true)
| | |-- element: struc
浏览 2提问于2017-09-26得票数 1
回答已采纳
1回答
显示.count()，但是.sum()出现错误，我该怎么办？ 代码： def meanTemperature(df,spark):
counttemp=spark.sql("SELECT temperature from washing").count()
sumtemp=spark.sql("SELECT temperature from washing").sum()
mean=sumtemp/counttemp
return mean 错误: AttributeError：'DataFrame‘对象没有'
浏览 45提问于2019-05-16得票数 0
2回答
这是dataframe模式结构。
root
|-- validFrom: array (nullable = true)
| |-- element: string (containsNull = true)
这就是我试图为validFrom获取日期数据类型的pyspark查询。
df_02 = spark.sql("""
select to_date(validFrom, 'yyyy-MM-dd') as validFrom
from v_source_df_flatten
""");
但我收到以下错误。
由
浏览 4提问于2022-01-02得票数 1
3回答
我正在尝试访问Tuple2中的一个字段，编译器正在返回一个错误。该软件试图在kafka主题中推送一个案例类，然后我希望使用火花流恢复它，这样我就可以提供机器学习算法，并将结果保存在mongo实例中。
解决了！
我终于解决了我的问题，我要发布最后的解决方案：
这是github项目：
https://github.com/alonsoir/awesome-recommendation-engine/tree/develop
build.sbt
name := "my-recommendation-spark-engine"
version := "1.0-SNAPSHOT
浏览 2提问于2016-05-18得票数 4
回答已采纳
下面是需要填充到dataframe中的数据
val columnNames = Array("ID", "Name","Age")
val d1 = Array("QWER","TOM","28")
val d2 = Array( "SPSRT","BENJI","45")
val d1zip = columnNames.zip(d1)
val d2zip1 = columnNames.zip(d2)
浏览 8提问于2022-03-25得票数 0
2回答
我有一个数据帧，其中有几个数值列是不固定的(它们在每次执行过程中都会发生变化)。假设我有一个带有数字列名称的Seq对象。我想对这些列中的每一列应用一个聚合函数。我尝试过以下几种方法：
println(numeric_cols)
// -> Seq[String] = List(avgTkts_P1, avgTkts_P2, avgTkts_P3, avgTkts_P4)
var sum_ops = for (c <- numeric_cols) yield org.apache.spark.sql.functions.sum(c).as(c)
var result = df.g
浏览 2提问于2018-09-04得票数 0
2回答
我有一个带有这种模式的JSON文件：
{
"name" : "john doe",
"phone-numbers" : {
"home": ["1111", "222"],
"country" : "England"
}
}
家庭电话号码数组有时可能是空的。
我的spark应用程序收到这些JSONS的列表，并执行以下操作：
val dataframe = spark.read.json(filePaths: _*)
val result = da
浏览 2提问于2017-05-21得票数 3
回答已采纳
我试图将一个列的数据main添加到更大的dataframe中，但是，第一个dataframe的问题是，在创建它之后，尝试通过命令将其添加到主dataframe中：
df.withColumn("name", dataframe)
我知道错误：
**found : org.apache.spark.sql.DataFrame
(which expands to) org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]
required: org.apache.spark.sql.Column**
我知道Data
浏览 0提问于2018-05-20得票数 3
回答已采纳
2回答
让我们假设拥有这个Pyspark dataframe：
x = np.random.randint(1, 100, 1000)
y = np.random.randint(1, 100, 1000)
z = np.random.randint(1, 100, 1000)
df = pd.DataFrame({'x': x, 'y': y, 'z': z})
spark_df = spark.createDataFrame(df)
让我们假设有以下值列表：
lst = [10, 20, 30]
我想检索所有3 (=len(lst))行的spark
浏览 5提问于2021-09-28得票数 4
回答已采纳
使用snowflake spark连接器从snowflake复制表时，默认行为是将结构化数据映射到spark字符串：https://docs.snowflake.net/manuals/user-guide/spark-connector-use.html#from-snowflake-to-spark-sql 例如，给定snowflake中的一个表： create table schema.table as
select
array_construct('1','a') as array_col,
object_construct(
浏览 26提问于2019-05-30得票数 1
我有一个json数据集，它的格式如下：
val data = spark.read.json("user.json").select("user_id","friends").show()
+--------------------+--------------------+
| user_id| friends|
+--------------------+--------------------+
|18kPq7GPye-YQ3LyK...|[rpOyqD_893cqmDAt...|
|rpOy
浏览 2提问于2017-07-04得票数 5
回答已采纳
我正在为Spark使用Java，并且想联合两个DataFrames，但是奇怪的是，DataFrame类只有unionAll？这是故意的吗?有没有办法在不复制的情况下将两个DataFrames联合起来？
浏览 1提问于2016-01-25得票数 7
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
