Spark scala在另一个数据帧中查找值
Spark Scala是一种基于Scala语言的开源分布式计算框架，用于处理大规模数据集的计算任务。它提供了丰富的API和功能，可以高效地进行数据处理、分析和机器学习等任务。
在另一个数据帧中查找值可以通过Spark Scala的DataFrame API来实现。DataFrame是一种分布式的数据集合，类似于关系型数据库中的表，可以进行类似SQL的查询操作。
要在另一个数据帧中查找值，可以使用DataFrame的filter函数结合条件表达式来实现。例如，假设有两个数据帧df1和df2，我们想要在df2中查找满足某个条件的值，可以使用如下代码：
val result = df2.filter(df2("column_name") === value)其中，column_name是df2中的列名，value是要查找的值。上述代码将返回一个新的数据帧result，其中包含满足条件的行。
除了filter函数，还可以使用其他DataFrame的函数来实现更复杂的查找操作，例如select、join等。
在腾讯云的云计算平台中，推荐使用TencentDB for Apache Spark作为Spark Scala的底层数据存储和计算引擎。TencentDB for Apache Spark是腾讯云提供的一种高性能、弹性扩展的云原生分析数据库服务，可以与Spark Scala无缝集成，提供稳定可靠的数据存储和计算能力。
更多关于TencentDB for Apache Spark的信息和产品介绍，可以访问腾讯云官方网站：TencentDB for Apache Spark
相关·内容
我正在尝试从我的Spark作业中写出一些元数据到postgres数据库。继续输入以下错误：
2018-03-13 15:23:40,599 root ERROR An error occurred while calling o1315.jdbc.
: scala.MatchError: null
at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:62)
at org.apache.spark.sql.
浏览 1提问于2018-03-14得票数 0
我试图在spark scala中关联两列数据帧，方法是将原始数据帧的列通过管道传输到VectorAssembler中，然后是关联实用程序。由于某些原因，Vector汇编程序似乎生成了空向量，如下所示。这是我到目前为止所掌握的。 val numericalCols = Array(
"price", "bedrooms", "bathrooms",
"sqft_living", "sqft_lot"
)
val data: DataFrame = HousingD
浏览 17提问于2020-07-29得票数 0
我正试着从Spark数据帧中写给Cassandra。当我有一个简单的数据帧模式时，如示例中所示，它可以工作：
root
|-- id: string (nullable = true)
|-- url: string (nullable = true)
但是，当我尝试编写包含StructTypes的数据帧时，其模式如下：
root
|-- crawl: struct (nullable = true)
| |-- id: string (nullable = true)
然后我得到以下异常：
Exception in thread "main" java.lang
浏览 4提问于2016-06-27得票数 1
几天后，数据带来了大量的功能。为了获得简短的数据摘要，人们将数据加载到数据框中，并使用head()方法显示数据。使用Jupyter Notebook( Toree for scala)进行实验是很常见的。
Spark (scala)很适合处理大量数据，但是它的head()方法不能在水平滚动的notebook中显示列标题。
熊猫数据帧头
Spark Scala数据帧头
我知道你可以通过使用.columns在scala dataframe中获得列标题，但是打印它不会沿着数据列显示标题，这使得它很难理解。
浏览 3提问于2018-06-13得票数 2
3回答
在spark (如 )中，有许多方法来验证两个数据帧的架构。但是我只想在SQL中验证两个数据帧的模式，我指的是SparkSQL。
示例查询1:
SELECT DISTINCT target_person FROM INFORMATION_SCHEMA.COLUMNS WHERE COLUMN_NAME IN ('columnA','ColumnB') AND TABLE_SCHEMA='ad_facebook'
示例查询2:
SELECT count(*) FROM information_schema.columns WHERE table_n
浏览 6提问于2018-09-04得票数 1
我对apache-spark有个问题。我使用的是yarn客户端模式。
这是我的配置：
conf.set("spark.executor.memory", "4g")
conf.set("spark.driver.memory", "4g")
conf.set("spark.driver.cores", "4")
conf.set("spark.default.parallelism", "3")
conf.set("spark.executor.cores&#
浏览 4提问于2016-01-06得票数 0
我在scala中有一个名为df的spark数据帧，它有两列，即a和b，a列包含字母，b列包含数字，如下所示。
a b
----------
g 0
f 0
g 0
f 1
我可以使用以下命令获取不同的行
val dfDistinct=df.select("a","b").distinct
它提供了以下内容：
a b
----------
g 0
f 0
f 1
我想添加另一个列，其中包含这些不同的组合在第一个数据帧中出现的次数，这样我就可以得到
a b count
-
浏览 2提问于2015-10-28得票数 9
回答已采纳
我使用的是Spark 1.6，正在尝试将csv (或tsv)文件读取为数据帧。以下是我采取的步骤：
scala> val sqlContext= new org.apache.spark.sql.SQLContext(sc)
scala> import sqlContext.implicits._
scala> val df = sqlContext.read
scala> .format("com.databricks.spark.csv")
scala> .option("header", "true")
sc
浏览 12提问于2016-07-27得票数 2
我有两个数据帧，第一个数据帧包含实际数据(从CSV文件读取)，第二个数据帧包含一个具有多个关键字的列。即下面的Dataframe-1外观相似(我们必须搜索的地方)：
数据帧-2：
我想要的输出：
我在这里使用spark scala。我想要一个与dataframe-1中的dataframe-2完全匹配的单词。我使用了like、rlike、contains等函数，但它没有给出我想要的输出。有人知道如何在spark scala SQL中或使用dataframe的spark scala函数来开发此逻辑吗？请帮我弄一下这个。
浏览 1提问于2021-02-12得票数 0
我正在尝试在spark 1.5.0中实现下面的oracle逻辑，我有如下select查询
例如：
select name, id, age, country, CASE WHEN (id is not null AND NVL(country,'DUMMY') NOT IN (us,'DUMMY') )
THEN correct
ELSE wrong
END Code(one of the column)
from employee
。
我想在spark中实现case和NVL逻辑，有人可以帮助我如何使用数据帧在spark scala API中实现这个逻辑吗？
浏览 1提问于2017-01-24得票数 1
我为ApacheSpark2.3.0编写了一个自定义的状态存储和状态存储提供程序，并尝试使用附加参数部署该作业：
--conf spark.sql.streaming.stateStore.providerClass=com.sample.state.CustomStateStoreProvider
对于运行星火作业，我使用马拉松和Mesos，该作业在异常开始后就会失败：
java.lang.ClassNotFoundException: com.sample.state.CustomStateStoreProvider
at java.net.URLClassLoader.find
浏览 0提问于2018-07-20得票数 5
回答已采纳
我有一个数据帧，如下所示，称为training +------------------+------+
| features| MEDV|
+------------------+------+
| [6.575,4.98,15.3]|504000|
| [6.421,9.14,17.8]|453600|
| [7.185,4.03,17.8]|728700|
| [6.998,2.94,18.7]|701400| 我对此数据集运行线性回归 from pyspark.ml.regression import LinearRegression
lr=LinearRegres
浏览 16提问于2020-06-15得票数 0
回答已采纳
下面是我的密码。
directKafkaStream.foreachRDD(rdd ->
{
rdd.foreach(record ->
{
messages1.add(record._2);
});
JavaRDD<String> lines = sc.parallelize(messages1);
JavaPairRDD<Integer, String> data = lines
浏览 1提问于2017-03-09得票数 0
回答已采纳
我正在使用sqlContext数据帧/ scala，并且已经成功地写出了其中的一些数据。现在我需要对另一个表做一个正确的联接，它只将它过滤成我感兴趣的数据。但是，当我试图编写这个连接表时，它会产生一个NullPointerException。
这样做是可行的：
data
.select($"id", $"text")
.map(x => (x.getString(0), x.getString(1)).productIterator.mkString("\t"))
.saveAsTextFile("/hdfs
浏览 1提问于2017-06-23得票数 0
回答已采纳
我用下面的代码得到了我的spark数据帧
scala> val sqlContext = new org.apache.spark.sql.SQLContext(sc)
scala> import sqlContext.implicits._
scala> case class Wiki(project: String, title: String, count: Int, byte_size: String)
scala> val data = sc.textFile("s3n://+++/").map(_.split(" ")).
浏览 1提问于2015-10-15得票数 0
我正在尝试找到将整个Spark数据帧转换为scala Map集合的最佳解决方案。最好的说明如下：
从这里开始(在Spark示例中)：
val df = sqlContext.read.json("examples/src/main/resources/people.json")
df.show
+----+-------+
| age| name|
+----+-------+
|null|Michael|
| 30| Andy|
| 19| Justin|
+----+-------+
添加到Scala集合(Map of Maps)，如下所示：
val peop
浏览 2提问于2016-04-28得票数 13
回答已采纳
2回答
我正在使用这个版本的Spark：spark-1.4.0-bin-hadoop2.6。我想检查几个默认属性。因此，我在spark-shell中给出了以下语句
scala> sqlContext.getConf("spark.sql.hive.metastore.version")
我期望对方法getConf的调用会返回此中描述的0.13.1值。但是我得到了下面的异常
java.util.NoSuchElementException: spark.sql.hive.metastore.version
at org.apache.spark.sql.SQLConf$$a
浏览 0提问于2015-07-17得票数 6
回答已采纳
我正在使用Datastax spark- Cassandra -connector来访问Cassandra中的一些数据。我的需求是将RDD与Cassandra表连接起来，获取结果并将其存储在hive表中。 我正在使用joinWithCassandraTable加入cassadra表。加入后，生成的RDD如下所示 com.datastax.spark.connector.rdd.CassandraJoinRDD[org.apache.spark.sql.Row,
com.datastax.spark.connector.CassandraRow] =
CassandraJoinRDD[17
浏览 20提问于2020-07-17得票数 0
我有两个不同类型的值，如下所示
scala> val ageSum = df.agg(sum("age"))
ageSum: org.apache.spark.sql.DataFrame = [sum(age): bigint]
scala> val totalEntries = df.count();
scala> totalEntries
res37: Long = 45211
第一个值来自数据帧上的聚合函数，第二个值来自数据帧上的总计数函数。两者都有不同的类型，因为ageSum是bigInt，totalEntries是长的。我想对它进行数学运算。平均值=
浏览 7提问于2017-01-26得票数 0
回答已采纳
我使用的是spark 2.3.2，在对来自Kafka的2个或更多流媒体资源进行联合时遇到了一个问题。这些都是来自Kafka的流媒体源，我已经将它们转换并存储在Dataframes中。 理想情况下，我希望将此UNIONed数据帧的结果以parquet格式存储在HDFS中，甚至可能存储回Kafka中。最终目标是以尽可能低的延迟存储这些合并的事件。 val finalDF = flatDF1
.union(flatDF2)
.union(flatDF3)
val query = finalDF.writeStream
.format("parquet&
浏览 52提问于2019-07-02得票数 1
回答已采纳
我试图在PySpark和笔记本中创建和分析数据。
下面是我在朱庇特笔记本上的代码。
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.master("local") \
.appName("Neural Network Model") \
.config("spark.executor.memory", "6gb") \
.getOrCreate()
我开始了火花疗法。
df1 = spark.createDataFr
浏览 0提问于2019-01-25得票数 1
我正在使用pyspark随机森林分类器，并喜欢在得到预测后根据它们创建一个熊猫数据帧。最奇怪的异常发生在我尝试这样做的时候。下面是我的代码：
random_forest = RandomForestClassifier(labelCol = 'label', featuresCol = 'features', maxDepth = 4, impurity = 'entropy', numTrees = 10, maxBins = 250)
rf_model = random_forest.fit(training_data)
prediction
浏览 2提问于2019-01-29得票数 0
我试图在数据帧中加载一个CSV文件，我的目标是将第一行显示为CSV文件的列名。但是在使用下面的代码时，我得到了错误
Exception in thread "main" java.lang.AbstractMethodError
at scala.collection.TraversableLike$class.filterNot(TraversableLike.scala:278)
代码：
def main(args : Array[String]): Unit = {
val spark : SparkSession = SparkSession
.b
浏览 0提问于2018-12-31得票数 2
回答已采纳
3回答
我来自python背景，试图将中的函数转换为scala。 在这个虚拟示例中，我有多个(未知数量)的数据帧需要合并在一起。 %python
list_of_dfs = [
spark.createDataFrame(
[('A', 'C'),
('B', 'E')
], ['dummy1','dummy2']),
spark.createDataFrame(
[('F', &#
浏览 24提问于2021-09-29得票数 2
回答已采纳
我有以下数据帧： dataframe1
+-----------------------+
|ID |
+-----------------------+
|[10,80,60,] |
|[20,40,] |
+-----------------------+ 和另一个数据帧： dataframe2
+------------------+----------------+
|ID_2 | name |
+------------------+-------
浏览 76提问于2020-08-15得票数 0
有一些函数可以删除Spark(SCALA)中的列和行，但是却找不到任何函数来删除整个数据frame.Is，有一种方法可以删除Spark(SCALA)中的数据帧吗？
浏览 0提问于2016-10-15得票数 3
我有数据帧df
| name| languagesAtSchool|currentState|
+----------------+------------------+------------+
| James,,Smith|[Java, Scala, C++]| CA|
| Michael,Rose,|[Spark, Java, C++]| NJ|
|Robert,,Williams| [CSharp, VB, R]| NV|
+----------------+------------------+
浏览 11提问于2021-10-11得票数 2
在我们的一个Pyspark作业中，我们有一个场景，我们在一个大的数据帧和相对较小的数据帧之间进行连接，我相信spark正在使用广播连接，我们遇到了以下错误 org.apache.spark.SparkException: Cannot broadcast the table that is larger than 8GB: 8 GB
at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec$$anonfun$relationFuture$1$$anonfun$apply$1.apply(BroadcastExch
浏览 752提问于2020-04-22得票数 0
我正在使用spark structured来编写一些使用function的转换数据帧：
def parquetStreamWriter(dataPath: String, checkpointPath: String)(df: DataFrame): Unit = {
df.writeStream
.trigger(Trigger.Once)
.format("parquet")
.option("checkpointLocation", checkpointPath)
.start(dataPath)
}
当我在代码中调用这个函数的次
浏览 7提问于2018-07-26得票数 1
3回答
我有一个包含多个列的spark数据帧。我想找出并删除列中有重复值的行(其他列可以是不同的)。
我尝试使用dropDuplicates(col_name)，但它只删除重复的条目，但仍然在数据帧中保留一条记录。我需要的是删除所有最初包含重复条目的条目。
我使用的是Spark 1.6和Scala 2.10。
浏览 4提问于2018-04-10得票数 5
回答已采纳
我正在尝试使用scala中的withColumn向spark数据帧中添加一列，
因为数据集很大，所以我的withColumn在特定的记录上失败了，
我正在尝试用try-catch编写这段代码，这样我就可以捕获异常并记录这些错误记录，然后继续剩余的记录。
在spark-scala中有没有一种方法来处理withColumn中发生的异常？
请提个建议。
浏览 2提问于2020-03-04得票数 0
我已经创建了一个函数，它接受一行作为输入，并给出一个字符串作为输出。我计划将此函数应用于不同模式的各种数据帧。这些数据帧非常庞大，每个数据帧有数百万行，但每个数据帧都有一个定义好的模式 我想创建另一个函数，它将调用第一个函数，将函数的输出字符串与它发送到函数的行合并，并创建一个新的数据帧，该数据帧将作为第二个函数的输出。 这两个函数都将在spark-scala环境中编写。我是spark-scala的新手，不太确定如何将行组合成一个新的数据帧 def returnTranformFunctionOutput(inputDataRow: Row, TransformFrame: Array[Ro
浏览 46提问于2019-10-15得票数 0
回答已采纳
在spark scala dataframe中定义多个逻辑/关系条件时，得到下面提到的错误。但同样的事情在scala中也能正常工作
Python代码：
df2=df1.where(((col('a')==col('b')) & (abs(col('c')) <= 1))
| ((col('a')==col('fin')) & ((col('b') <= 3) & (col('c') > 1)) & (col('d')
浏览 12提问于2019-08-02得票数 0
我正在尝试将spark数据帧加载到Cassandra表中，该数据帧具有两个具有集合数据类型的属性。 在传入的提要文件中，这些属性是文本/字符串。我使用下面的代码将String类型分别转换为List和Map类型： spark.udf.register("getLst", (input: String) => input.split(",").toList)
spark.udf.register("getMap", (input:String) => parse(input).values.asInstanceOf[M
浏览 18提问于2019-01-23得票数 0
回答已采纳
我正在尝试修复在我的spark设置中看到的一个外存问题，在这一点上，我无法得出一个具体的分析，为什么我会看到这个。在将数据帧写到parquet或kafka时，我总是会遇到这个问题。我的数据帧有5000行。它的模式是
root
|-- A: string (nullable = true)
|-- B: string (nullable = true)
|-- C: string (nullable = true)
|-- D: array (nullable = true)
| |-- element: string (contains
浏览 0提问于2017-09-14得票数 0
我在Spark中有以下数据框架和模式
val df = spark.read.options(Map("header"-> "true")).csv("path")
scala> df show()
+-------+-------+-----+
| user| topic| hits|
+-------+-------+-----+
| om| scala| 120|
| daniel| spark| 80|
|3754978| spark| 1|
+-------+-------+-----+
浏览 118提问于2017-06-21得票数 1
我使用StructType定义了一个模式，用于读取Redsfhit中的数据帧。该表具有350+列，其中许多列被配置为布尔值。
从该表中卸载数据后，我尝试使用我创建的Schema读取数据。但是我希望读取的每个BooleanType列都有"f“/ "t”值。这会导致解析异常。例外情况是：
java.lang.IllegalArgumentException: For input string: "f"
at scala.collection.immutable.StringLike$class.parseBoolean(StringLike.scala:290)
a
浏览 1提问于2017-11-22得票数 3
我有一个名为rankedDF的数据文件：
+----------+-----------+----------+-------------+-----------------------+
|TimePeriod|TPStartDate| TPEndDate|TXN_HEADER_ID| Items|
+----------+-----------+----------+-------------+-----------------------+
| 1| 2017-03-01|2017-05-30| TxnHeader1|Womens Soc
浏览 0提问于2018-04-24得票数 0
回答已采纳
我正在尝试加入两个数据帧，一个大约是1000万条记录，另一个大约是其中的1/3。由于小DataFrame非常适合于执行器的内存，因此我执行了一个广播连接，然后写出结果：
val df = spark.read.parquet("/plablo/data/tweets10M")
.select("id", "content", "lat", "lon", "date")
val fullResult = FilterAndClean.performFilter(df, spark)
.sele
浏览 5提问于2017-11-21得票数 0
我有以下问题。我希望将变量(字符串)的值从scala传递给databricks中的python。我知道我可以使用以下命令在两种语言的之间传输数据帧信息： %scala
scalaDF.registerTempTable("some_table")
%python
spark.table("some_table") 但是我不能用这种方式传输字符串，你有什么想法吗？
浏览 19提问于2021-01-26得票数 1
1回答
假设我下面有两个数据帧，这两个数据帧将通过scala代码创建，然后jar将被构建。
{
def main(args: Array[String])
{
val df1 = sc.parallelize(Seq(1,0,1,0)).toDF("col1")
val df2 = sc.parallelize(Seq(234,43,341,42)).toDF("col2")
}
}
我的要求是创建jar并提交spark提交。
假设dataframe 1由于某些问题而失败，submit应该退出而不继续dataframe df2。
如何处理这个错
浏览 0提问于2019-11-22得票数 0
我在HDFS中有一个csv文件，并且正在尝试将它加载到一个中，在EMR中使用pyspark一个python脚本。
我得到以下错误(最后是完全错误)
py4j.protocol.Py4JJavaError: An error occurred while calling o32.csv
下面是我要做的事情
df = spark.read.csv("http://localhost:9870/foo/tsla_202210_min.csv", schema = stockSchema)
我把什么东西放错了吗？
全误差
File "/home/hadoop/.local/
浏览 9提问于2022-11-20得票数 0
我正在尝试将Spark数据帧写入Hbase，但当我在同一数据帧上执行任何操作或写入/保存方法时，它会给出以下异常： {
java.lang.AbstractMethodError
at org.apache.spark.Logging$class.log(Logging.scala:50)
at org.apache.spark.sql.execution.datasources.hbase.HBaseFilter$.log(HBaseFilter.scala:121)
at org.apache.spark.sql.execution.dat
浏览 11提问于2019-02-21得票数 1
Spark版本: 1.3
在处理BigInteger类型数据时，我有一个要求。Bean类(Pojo)使用的BigInteger数据类型很少。解析数据并创建JavaRDD可以很好地工作，但是当创建一个以JavaRDD和BeanClass为参数的数据帧时，Spark抛出下面的异常。
scala.MatchError: class java.math.BigInteger (of class java.lang.Class)
at org.apache.spark.sql.SQLContext$$anonfun$getSchema$1.apply(SQLContext.scala:1
浏览 3提问于2015-11-19得票数 1
我对scala中spark的数据帧有一个问题。我使用方法var df = spark.read.format("csv").load("csvfile.csv")来读取CSV文件并将其存储在DF中。我的CSV文件中有16行我不想读的注释。我还没有找到避免头部的方法，但它只有一行。有什么想法吗？
谢谢你。
浏览 4提问于2018-08-17得票数 0
1回答
我正在尝试过滤一个数据帧与另一个数据帧：
scala> val df1 = sc.parallelize((1 to 100).map(a=>(s"user $a", a*0.123, a))).toDF("name", "score", "user_id")
scala> val df2 = sc.parallelize(List(2,3,4,5,6)).toDF("valid_id")
现在我想过滤df1并得到一个数据帧，其中包含df1中的所有行，其中user_id在df2("vali
浏览 2提问于2015-09-19得票数 15
回答已采纳
环境- spark-3.0.1-bin-hadoop2.7、eclipse 2.12.3、Scala、SparkSQL、eclipse-jee-oxygen-2-linux-gtk-x86_64
我有一个csv文件，它有3列数据类型:String，Long，Date。我已经将csv文件转换为数据帧，并想要显示它。但是它给出了以下错误
java.lang.ArrayIndexOutOfBoundsException: 2
at org.apache.spark.examples.sql.SparkSQLExample5$.$anonfun$runInferSchemaExample$2(Spar
浏览 1提问于2021-03-25得票数 0
我在将spark数据帧写入csv和parquet时出错。我已经尝试安装winutil，但是仍然不能解决这个错误。
我的代码
INVALID_IMEI <- c("012345678901230","000000000000000")
setwd("D:/Revas/Jatim Old")
fileList <- list.files()
cdrSchema <- structType(structField("date","string"),
浏览 0提问于2017-09-23得票数 0
我正在使用下面的内容创建一个使用hive外部表的dataframe (spark scala)。但数据帧也在其中加载了数据。我需要一个使用配置单元外部表的模式创建的空DF。我使用spark scala来解决这个问题。 val table1 = sqlContext.table("db.table") 如何使用配置单元外部配置单元表创建空数据帧？
浏览 20提问于2019-10-16得票数 1
如何复制此代码以获取pyspark中的数据帧大小？
scala> val df = spark.range(10)
scala> print(spark.sessionState.executePlan(df.queryExecution.logical).optimizedPlan.stats)
Statistics(sizeInBytes=80.0 B, hints=none)
我想要做的是将sizeInBytes值放入一个变量中。
浏览 0提问于2020-06-03得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
