Update with inner join using spark dataframe/dataset/RDD
在Spark中使用DataFrame/Dataset/RDD进行内连接的更新操作，可以通过以下步骤实现：
- 首先，将需要更新的数据加载到一个DataFrame/Dataset/RDD中，假设为updateData。
- 然后，将需要更新的数据与目标表进行内连接操作，得到一个新的DataFrame/Dataset/RDD，假设为joinedData。内连接可以使用join函数，并指定连接条件。
- 接下来，使用withColumn函数将需要更新的列添加到joinedData中。可以使用when和otherwise函数来根据条件进行更新。例如，可以使用when(col("condition"), col("new_value")).otherwise(col("old_value"))来更新列的值。
- 最后，将更新后的数据写入到目标表中。可以使用write函数将DataFrame/Dataset/RDD写入到目标表中。
以下是一个示例代码，演示如何使用Spark DataFrame进行内连接的更新操作：
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
// 创建SparkSession
val spark = SparkSession.builder()
.appName("Update with inner join using Spark DataFrame")
.getOrCreate()
// 加载需要更新的数据
val updateData = spark.read.format("csv").load("path_to_update_data.csv")
// 加载目标表数据
val targetTable = spark.read.format("csv").load("path_to_target_table.csv")
// 执行内连接操作
val joinedData = targetTable.join(updateData, targetTable("join_column") === updateData("join_column"), "inner")
// 添加需要更新的列
val updatedData = joinedData.withColumn("updated_column", when(col("condition"), col("new_value")).otherwise(col("old_value")))
// 将更新后的数据写入目标表
updatedData.write.format("csv").save("path_to_output_table.csv")在上述示例中，需要将"path_to_update_data.csv"和"path_to_target_table.csv"替换为实际的数据文件路径。同时，需要根据实际情况修改连接条件、更新列的条件和值，以及输出表的路径和格式。
请注意，这只是一个示例代码，具体的实现方式可能因使用的Spark版本和具体需求而有所不同。此外，还可以根据具体情况选择使用DataFrame、Dataset或RDD来进行操作。
相关·内容
3回答
斯帕克-Scala中将Dataset[Row]转换为RDD[Array[String]]的最佳方法？
scala、apache-spark、apache-spark-sql、rdd、apache-spark-dataset
我正在创建一个火花数据集通过读取一个csv文件。此外，我需要将这个DatasetRow转换为RDD[ArrayString]，以便将它传递给FpGrowth(Spark )。
val df: DataFrame = spark.read.format("csv").option("header", "true").load("/path/to/csv")
val ds: Dataset[Row] = df.groupBy("user").agg(collect_set("values"))
现在，我
浏览 8提问于2021-01-08得票数 3
回答已采纳
5回答
更新spark中的dataframe列
python、dataframe、apache-spark、pyspark、apache-spark-sql
查看新的spark API，还不清楚是否有可能修改DataFrame列。
如何更改数据文件的行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)
浏览 12提问于2015-03-17得票数 95
回答已采纳
1回答
我见过很多这样的关于序列化错误的帖子。但我对此还不熟悉。
这里有一个dataframe-modProductsData和一个地图L2L3Map地图。我想用map-L2L3Map的值替换column-PRIMARY_CATEGORY中的值。
val L2L3Map = L2.collect.map(row => (row.get(0).toString, row.get(1).toString)).toMap
val L2L3MapUDF = udf { s: String => L2L3Map.get(s) }
val productsData = spark.read.forma
浏览 2提问于2018-12-21得票数 0
1回答
转换为RDD失败
scala、entity-framework、apache-spark
我的代码如下。我读取了一个包含两列的CSV文件。通过转换为RDD循环访问Dataframe的元素。现在我想为每个元素创建一个DF。下面的代码失败。有谁能帮帮忙。 val df1 = spark.read.format("csv").load("c:\\file.csv") //CSV has 3 columns
for (row <- df1.rdd.collect)
{
var tab1 = row.mkString(",").split(",")(0) //Has
浏览 23提问于2021-09-06得票数 0
回答已采纳
1回答
写入数据集/数据集时出错。如何正确创建表火花？
scala、apache-spark、apache-spark-sql、apache-spark-dataset
在对原始数据集/dataframe执行几个转换(联合)后，我希望将dataset/dataframe写入csv。获得的dataset/dataframe显示得很完美，没有任何问题，但是当我想在csv中编写dataset/dataframe时，它会显示以下错误
Caused by: java.io.FileNotFoundException: File file:/../file.csv does not exist
It is possible the underlying files have been updated. You can explicitly invalidate the c
浏览 17提问于2022-06-13得票数 1
1回答
Scala/Apache火花转换DataFrame列值和类型，否则为多次
sql、scala、apache-spark、cassandra
我有一个主SQL表，我正在读取这个表，并修改它以写入CassandraDB。目前，我有一个工作实现，将性别从0、1、2、3(整数)转换为“男性”、“女性”、“跨”等(字符串)。尽管下面的方法确实有效，但将带有这些映射的独立数组添加到DataFrame中、将其连接到主表/DataFrame、然后删除、重命名等等，似乎效率很低。
我见过：
.withColumn("gender", when(col("gender) === 1, "male").otherwise("female")
这将允许我继续在主表上进行方法链接，但无法让它使用两个
浏览 1提问于2017-09-11得票数 2
回答已采纳
1回答
数据库中csv格式到Redis散列格式的数据传输
scala、apache-spark、redis、databricks、azure-databricks
我有一个Azure系统，分为三个部分：
Azure数据湖存储，在那里我有一些csv文件。
，我需要做一些处理-确切地说是将csv文件转换为Redis散列格式。
Azure Redis缓存，我应该把转换的数据放在哪里。
在databricks文件系统中安装存储后，需要处理一些数据。如何将位于databricks文件系统中的csv数据转换为redisHash格式，并正确地将其转换为Redis？具体来说，我不知道如何通过下面的代码进行正确的映射。或者，是否有其他方法可以转移到SQL表，而我找不到。
下面是我在scala上编写的代码示例：
import com.redislabs.
浏览 11提问于2020-11-09得票数 1
回答已采纳
1回答
从另一个数据库导入到Spark 2中的对象的类型是什么？
scala、apache-spark、apache-spark-sql、spark-dataframe
我把甲骨文中的两张桌子像这样导入到火花中：
val customers = spark.read.format("jdbc").option("url", "jdbc:oracle:thin:user/password@//IP:2222/ebe").option("driver", "oracle.jdbc.OracleDriver").option("dbtable", "customers").load().show()
val cities = spark.read.for
浏览 0提问于2018-02-15得票数 1
回答已采纳
3回答
Spark Dataframe嵌套Case When语句
sql、apache-spark、dataframe、apache-spark-sql
我需要在Spark DataFrame中实现以下SQL逻辑。
SELECT KEY,
CASE WHEN tc in ('a','b') THEN 'Y'
WHEN tc in ('a') AND amt > 0 THEN 'N'
ELSE NULL END REASON,
FROM dataset1;
我的输入DataFrame如下：
val dataset1 = Seq((66, "a", "4"), (67, "a
浏览 2提问于2017-10-09得票数 11
回答已采纳
3回答
带点火花的列名
scala、apache-spark、apache-spark-sql、apache-spark-mllib、apache-spark-ml
我正在尝试从DataFrame中获取列，并将其转换为RDD[Vector]。
问题是，我有一些列的名称中有一个“点”，如下面的数据集：
"col0.1","col1.2","col2.3","col3.4"
1,2,3,4
10,12,15,3
1,12,10,5
这就是我要做的：
val df = spark.read.format("csv").options(Map("header" -> "true", "inferSchema" -> "
浏览 1提问于2017-06-05得票数 30
3回答
基于第一个DataFrame Java中的列，使用新的列值创建一个新的Spark dataframe
java、apache-spark、dataframe、apache-spark-sql
这应该是easy but....using Spark 1.6.1....我有包含A、B、C列的DataFrame #1，其值如下：
A B C
1 2 A
2 2 A
3 2 B
4 2 C
然后，我用一个新的列D创建一个新的数据帧，因此：
DataFrame df2 = df1.withColumn("D", df1.col("C"));
到目前为止还不错，但我实际上希望D列中的值是有条件的，即：
// pseudo code
if (col C = "A") the col D = "X"
else if
浏览 3提问于2016-05-07得票数 12
回答已采纳
2回答
在S3中将每个分区数据写入单个文件中
scala、apache-spark
我们有这样的用例，我们希望按列值对数据帧进行分区，然后将每个分区写到单个文件中。我做了以下事情来做同样的事情：
val df = spark.read.format("csv").load("hdfs:///tmp/PartitionKeyedDataset.csv")
df.repartition($"_c1")
df.rdd.saveAsTextFile("s3://dfdf/test1234")
当我这样做的时候：
df.rdd.partitions.size
我只得到了62个partition.But，列的不同值是1
浏览 1提问于2018-03-14得票数 1
1回答
更好的方法是创建单独的Dataframe或引用同名的dataframe
apache-spark、pyspark
日安。
我是个新手，我希望得到一个更好的建议。每次完成新的转换时创建一个新的dataframe。转换后使用与以前的dataframe名称相同的名称
我已经讨论过其他堆栈溢出问题以及rdd和dataframe指南，但我无法找到解决方案。
备选案文1：
df=spark.read.format("csv").option("header", "true").load("csvfile.csv")
df1=df.filter(col('country')=='India')
df2=df1.groupBy
浏览 0提问于2019-04-22得票数 1
回答已采纳
1回答
行间DeepDiff
python、pandas
我有这个数据框架
import pandas as pd
from deepdiff import DeepDiff
df = pd.DataFrame({'col_new': ['a','b','c,d','d'],
'col_old': ['a','e','d','d'],
'col_val': [True,False,False,True]
浏览 12提问于2020-08-26得票数 1
1回答
基于条件创建火花数据框架
scala、apache-spark、hive、spark-dataframe
我有两个数据帧：dataframe1有70000行，如下所示：
location_id, location, flag
1,Canada,active
2,Paris,active
3,London,active
4,Berlin,active
第二，df lookup为每个位置修改了is (此数据帧不时被修改)，如下所示：
id,location
1,Canada
10,Paris
4,Berlin
3,London
我的问题是，我需要从location_id中获取新id作为lookup，如果location_id与id不同，那么，将相应位置的旧id保持为非活动id (用于维护历史数据)，而
浏览 1提问于2017-09-29得票数 0
回答已采纳
1回答
在更新表之后，像数组一样插入数据
postgresql
我有two tables
Table A
col1 col2 col3
Table B
table_name column_name new_value old_value
如果在table A上发生任何更新，它将在table B上插入数据
表B is==>的输出
table_name column_name new_value old_value
---------------- ------------------ -----------
浏览 3提问于2013-09-19得票数 2
1回答
火花误差-批量解析达到最大迭代(100)
apache-spark、apache-spark-sql、data-science
我正在研究Spark，在这里我需要找出两个大型CSV之间的区别。
迪夫应给予：
插入行或新记录/比较仅Id的更改的行(不包括插入的行)-比较所有列值已删除的行//只比较Id的
Spark 2.4.4 + Java
我正在使用Databricks来读/写CSV
Dataset<Row> insertedDf = newDf_temp.join(oldDf_temp,oldDf_temp.col(key)
.equalTo(newDf_temp.col(key)),"left_anti");
Long insertedCount = in
浏览 0提问于2020-04-09得票数 0
回答已采纳
2回答
如何在使用schema Spark读取csv时删除格式错误的行？
scala、csv、apache-spark、apache-spark-dataset
当我使用Spark DataSet加载csv文件时。我更喜欢清晰地指定模式。但是我发现有一些行与我的模式不兼容。列应该是双精度的，但有些行是非数字值。是否可以很容易地从DataSet中过滤出所有不符合我的模式的行？
val schema = StructType(StructField("col", DataTypes.DoubleType) :: Nil)
val ds = spark.read.format("csv").option("delimiter", "\t").schema(schema).load("f.
浏览 4提问于2018-04-09得票数 9
回答已采纳
2回答
无法将数据直接从Spark写入/保存到Ignite
java、scala、apache-spark、jdbc、ignite
我试着用jdbc编写数据文件来点燃，
星火版本是: 2.1
点火版本:2.3
JDK:1.8
Scala:2.11.8
这是我的代码片段：
def WriteToIgnite(hiveDF:DataFrame,targetTable:String):Unit = {
val conn = DataSource.conn
var psmt:PreparedStatement = null
try {
OperationIgniteUtil.deleteIgniteData(conn,targetTable)
hiveDF.foreachPartition({
浏览 4提问于2017-11-03得票数 1
2回答
Spark Java编辑列中的数据
java、apache-spark、apache-spark-sql、apache-spark-dataset
我想遍历spark DataFrame中列的内容，如果满足特定条件，则更正单元格中的数据 +-------------+
|column_title |
+-------------+
+-----+
|null |
+-----+
+-----+
|0 |
+-----+
+-----+
|1 |
+-----+ 假设我想要在列的值为null时显示其他内容，我尝试使用 Column.when() DataSet.withColumn() 但是我找不到正确的方法，我认为没有必要转换成RDD并遍历它。
浏览 44提问于2019-02-13得票数 4
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
