如何在不使用UDF的情况下在Scala Spark中将字符串列追加到数组字符串列？
在Scala Spark中，如果不使用UDF（User Defined Function），可以使用内置的函数和操作符来将字符串列追加到数组字符串列。
首先，假设我们有一个DataFrame，其中包含一个字符串列和一个数组字符串列。我们想要将字符串列的值追加到数组字符串列中。
import org.apache.spark.sql.functions._
// 创建示例DataFrame
val df = Seq(
("value1", Array("array1", "array2")),
("value2", Array("array3", "array4"))
).toDF("stringCol", "arrayCol")
// 使用concat函数将字符串列追加到数组字符串列
val result = df.withColumn("arrayCol", concat($"arrayCol", array($"stringCol")))
result.show(false)这将输出以下结果：
+---------+-------------------+
|stringCol|arrayCol |
+---------+-------------------+
|value1 |[array1, array2, value1]|
|value2 |[array3, array4, value2]|
+---------+-------------------+在上述代码中，我们使用了concat函数和array函数。concat函数用于将两个数组合并，而array函数用于将字符串列转换为单元素数组。通过将arrayCol和array(stringCol)传递给concat函数，我们将字符串列的值追加到数组字符串列中。
这是一个简单的方法来实现在Scala Spark中将字符串列追加到数组字符串列的操作，而不使用UDF。
相关·内容
因此，我想创建一个可以在Pyspark中使用的scala UDF。我想要的是接受一个字符串列表作为x，一个字符串列表作为y，并获得所有的字符串组合 因此，如果我有x= "a"，"B“和y="A"，"b”，我期望输出= [a，A，a，B，b，A，b，B] 我编写的Scala代码非常简单 (x: Seq[String], y: Seq[String]) => {for (a <- x; b <-y) yield (a,b)} 我已经创建了一个scala UDF来做这件事。它适用于Scala Spark。 我的问题是试图让这个在py
浏览 29提问于2020-10-09得票数 4
回答已采纳
2回答
我想做一件简单的事。我想把所有的事件统计成2分钟的时间戳。
效果很好。
df = df.groupBy(window(df["time_value"], "2 minutes")).count()
df.show()
+--------------------+-----+
| window|count|
+--------------------+-----+
|[2018-04-10 15:00...| 770|
|[2018-04-10 00:42...| 100|
|[2018-04-10 04:14...| 54|
|[2018-04-06 15:
浏览 1提问于2018-05-15得票数 1
回答已采纳
我正在努力创建一个udf来提取一些列数据。列是棘手的，因为有时它的字符串，但在许多情况下可以是结构。我只想考虑列是struct的时间，并为它提取所需的信息。
假设这个例子：
SELECT annoyingCol.data From SomeDf
annoyingCol.data等于字符串或结构，以避免出现像这样的错误：need struct type but got string;。我想知道是否可以创建一个检查列类型的udf，例如：
SELECT
case when isStruct(annoyingCol.data) then annoyingCol.data.my_data else
浏览 0提问于2018-05-23得票数 2
下面的代码将导致NullPointerException。我不确定，这个异常是发生在某些行上，还是总是因为dataframe很大而无法指向行。
def removeUnwantedLetters(str: String): String = {
str.split("\\W+").filter(word => (word.matches("[a-z]+") && (word.length > 1))).mkString(" ")
}
val myudf = spark.udf.register("le
浏览 0提问于2018-11-30得票数 0
回答已采纳
我已经编写了一个模块，其中包含了在PySpark DataFrames上工作的函数。它们对DataFrame中的列进行转换，然后返回一个新的DataFrame。下面是代码的一个示例，缩短为只包含其中一个函数：
from pyspark.sql import functions as F
from pyspark.sql import types as t
import pandas as pd
import numpy as np
metadta=pd.DataFrame(pd.read_csv("metadata.csv")) # this contains metad
浏览 6提问于2016-03-10得票数 7
回答已采纳
1回答
尝试使用以下命令检查某些列中是否有NaN值 ddf_temp = ddf.select('col1', 'col2' ...) # all int type
ddf_temp.select([count(when(isnull(c), c)).alias(c) for c in ddf_temp.columns]).show() 我可以找出哪些列给了我这些错误，但我找不到为什么会出现这样的错误： ---------------------------------------------------------------------------
Py4JJav
浏览 27提问于2019-08-23得票数 0
回答已采纳
1回答
我在AWS上使用Spacy预训练的大型模型生成的文档向量时遇到了非常糟糕的时间。当我将文档向量放入一个dataframe中时，问题就开始了。
例如，这段代码对我来说很好：
# Load infrastructure libraries
import pandas as pd
import numpy as np
# Load NLP libraries and tools
import spacy
# Prepare the Spacy NLP parsers
nlp = spacy.load('en_core_web_lg')
# Load Spark
from pys
浏览 9提问于2022-09-19得票数 2
1回答
我希望在用户设计的函数中使用位于另一个类中的方法，但它不起作用。
我有一个方法：
def traitementDataFrameEleve(sc:SparkSession, dfRedis:DataFrame, domainMail:String, dir:String):Boolean ={
def loginUDF = udf((sn: String, givenName:String) => {
LoginClass.GenerateloginPersone(sn,givenName,dfr)
})
dfEleve.
浏览 0提问于2017-04-25得票数 0
对于吡火花中的DataFrame，如果使用F.lit(1) (或任何其他值)初始化列，则将其赋值给pandas_udf内部的一些值(在本例中使用shift()，但可能发生在任何其他函数)，这将导致“值在索引上为空”错误。
有人能提供一些提示，为什么会发生这种情况？是火星雨里的虫子吗？
请参阅下面的代码和错误。
spark = SparkSession.builder.appName('test').getOrCreate()
df = spark.createDataFrame([Row(id=1, name='a', c=3),
Row(id=2, name=
浏览 0提问于2019-07-22得票数 4
回答已采纳
这个问题已经被问到了Scala的，但它对我没有帮助，因为我正在使用Java API。我真的把所有的东西和厨房的水槽都扔进去了，所以这是我的方法：
List<String> sourceClasses = new ArrayList<String>();
//Add elements
List<String> targetClasses = new ArrayList<String>();
//Add elements
dataset = dataset.withColumn("Transformer", callUDF(
浏览 40提问于2019-11-25得票数 1
回答已采纳
我试图根据现有列向我的数据框架添加一个新列，其中数据是一个自定义对象数组。假设对象类型为MyObject，我试图执行如下操作：
Column col = df.col("old_col");
Encoder<MyObject[]> encoder = Encoders.bean(MyObject[].class);
TypedColumn<Object, MyObject[]> typedColumn = col.as(encoder);
df = df.withColumn("new_col",functions.callUDF(
浏览 5提问于2017-09-19得票数 0
回答已采纳
我已经在Scala中创建了一个UDF，当我尝试用函数名注册这个UDF时，它显示了错误。
不工作
def IPConvertUDF = spark.udf.register("IPConvertUDF", IPConvert)
误差
error: missing argument list for method IPConvert
Unapplied methods are only converted to functions when a function type is expected.
You can make this conversion explicit by w
浏览 3提问于2019-07-15得票数 4
回答已采纳
1回答
我使用的是python2.7和spark版本2.2.0。我在pyspark中创建了一个数据框，它具有字符串列类型并包含URL。 df = spark.createDataFrame([('example.com?title=%D0%BF%D1%80%D0%B0%D0%B2%D0%BE%D0%B2%D0%B0%D1%8F+%D0%B7%D0%B0%D1%89%D0%B8%D1%82%D0%B0',)], ['url'])
df.show(1, False)
+---------------------------------------------------
浏览 11提问于2019-03-01得票数 0
回答已采纳
我的星火数据集中有一组动态列。我想传递列数组，而不是单独的列。我们如何编写UDF函数，以便它接受列数组。
我试过传递字符串序列，但它失败了。
static UDF1<Seq<String>, String> udf = new UDF1<Seq<String>, String>() {
@Override
public String call(Seq<String> t1) throws Exception {
return t1.toString();
浏览 0提问于2019-07-03得票数 1
1回答
我有一个返回数据的UDF。就像下面的那个
scala> predict_churn(Vectors.dense(2.0,1.0,0.0,3.0,4.0,4.0,0.0,4.0,5.0,2.0))
res3: org.apache.spark.sql.DataFrame = [noprob: string, yesprob: string, pred: string]
scala> predict_churn(Vectors.dense(2.0,1.0,0.0,3.0,4.0,4.0,0.0,4.0,5.0,2.0)).show
+------------------+------
浏览 3提问于2016-12-20得票数 3
回答已采纳
我正在创建一个UDF，它将为我找到每周的第一天。
UDF的输入将是来自在yyyy-MM-dd hh:MM:ss中存储日期时间的Dataframe的字符串列。
我同意可以在没有UDF的情况下建立同样的机制，但我想探索一下，所有这样做的选择。到目前为止，我通过UDF被困在了实现中。
重要备注--周开始日是星期一。
密码-
import org.apache.spark.sql.functions._
import java.time.format.DateTimeFormatter
import java.time.LocalDate
import org.joda.time.DateTimeCo
浏览 3提问于2020-07-02得票数 1
回答已采纳
2回答
我有一个dataframe，我想根据函数返回的值添加一个新列。这个函数的参数是来自同一个dataframe的四列。 one和 one有点类似于我想要的内容，但没有回答我的问题。
这是我的数据框架(列比这四列更多)
+ ------ + ------ + ------ + ------ +
| lat1 | lng1 | lat2 | lng2 |
+ ------ + ------ + ------ + ------ +
| -32.92 | 151.80 | -32.89 | 151.71 |
| -32.92 | 151.80 | -32.89 | 151.71
浏览 0提问于2018-11-22得票数 5
回答已采纳
我一直在尝试使用Quantlib和Spark，尝试在Pyspark中传递Quantlib函数参见下面的示例：
from QuantLib import *
from pyspark.sql.types import StringType
from pyspark.sql.functions import udf
df = sc.parallelize([("2016-10-01",),
("2016-11-01",),
("2016-12-01",)]).to
浏览 1提问于2017-02-15得票数 0
回答已采纳
我正在使用一个UDF函数应用于spark dataframe中的一个字符串列，该数据帧迭代一个words单词集，并查找给定的列字符串是否包含该集合中的任何单词(见下文)：
udf { (s: String) => words.value.exists(word => s.contains(word)) }
我需要如何修改这个函数，使其返回在字符串中找到的words集合中的所有项的列表？
我尝试过使用when和otherwise
udf { (s: String) => when(words.value.exists(word => s.contains(word)),
浏览 6提问于2017-08-25得票数 0
回答已采纳
我使用Spark2.4已经有一段时间了，最近几天我刚刚开始使用Spark3.0。在切换到Spark3.0运行udf((x: Int) => x, IntegerType)后，我得到了这个错误
Caused by: org.apache.spark.sql.AnalysisException: You're using untyped Scala UDF, which does not have the input type information. Spark may blindly pass null to the Scala closure with primitive-ty
浏览 9提问于2020-12-03得票数 6
回答已采纳
new是一个如下所示的rdd：
[('hotel stinsen', '59.066', '15.1115'),
('ankaret hotel and restaurant', '58.6725', '17.0975'),
('quality hotel ekoxen', '58.40574', '15.62391'),
('hotel nordica', '63.8532', '15.5652'
浏览 8提问于2022-10-08得票数 -1
回答已采纳
我有两个Scala代码-- MyMain.scala和MyFunction.scala，分别构建和构建的MyFunction jar将在MyMain中充当UDF。
MyFunction.scala基本上包含一个带有公共方法public String myFunc(String val0, String val1)的Java类。该项目是在SBT中构建的，build_jar编译输出作为工件存储(只存储所需的类，即MyFunction.class，而不是依赖项)。
MyMain.scala将上述工件jar导入lib文件夹下，并使用unmanagedBase := baseDirectory.valu
浏览 1提问于2019-10-27得票数 2
回答已采纳
3回答
Spark作业在纱线模式下失败
、、、、
我有一个用Scala编写的Spark程序，它从HDFS读取CSV文件，计算新列，并将其另存为parquet文件。我在一个纱线集群中运行这个程序。但每次我尝试启动它时，executors都会在某些时候失败，并显示以下错误。
你能帮我找出可能导致这个错误的原因吗？
从executor登录
16/10/27 15:58:10 WARN storage.BlockManager: Putting block rdd_12_225 failed due to an exception
16/10/27 15:58:10 WARN storage.BlockManager: Block rdd_12_22
浏览 0提问于2016-10-28得票数 3
我在Scala上使用Spark 2.1。
如何将毫秒字符串列转换为毫秒时间戳？
我尝试了问题中的以下代码
import org.apache.spark.sql.functions.unix_timestamp
val tdf = Seq((1L, "05/26/2016 01:01:01.601"), (2L, "#$@#@#")).toDF("id", "dts")
val tts = unix_timestamp($"dts", "MM/dd/yyyy HH:mm:ss.SSS").cast(
浏览 24提问于2017-07-03得票数 9
回答已采纳
我正在开发一个spark 2.0.0版本，其中我的需求是在我的sql上下文中使用'com.facebook.hive.udf.UDFNumberRows‘函数来使用其中一个查询。在我的集群with Hive查询中，我将其用作临时函数，只需定义:创建临时函数myFunc为'com.facebook.hive.udf.UDFNumberRows'，这非常简单。
我尝试将其注册到sparkSession，如下所示，但得到一个错误：
sparkSession.sql("""CREATE TEMPORARY FUNCTION myFunc AS '
浏览 1提问于2016-11-02得票数 5
我正在尝试将一个word2vec模型对象传递给我的spark udf。基本上，我有一个带有电影I的测试集，我希望将这些I与model对象一起传递，以获得每一行的推荐电影数组。
def udfGetSynonyms(model: org.apache.spark.ml.feature.Word2VecModel) =
udf((col : String) => {
model.findSynonymsArray("20", 1)
})
然而，这给了我一个空指针异常。当我在udf之外运行model.findSynonymsArray("
浏览 3提问于2018-04-27得票数 3
2回答
我有一个字符串列表，这些字符串表示数据帧中的列名。我想将这些列中的参数传递给udf。我如何在spark scala中做到这一点？
val actualDF = Seq(
("beatles", "help|hey jude","sad",4),
("romeo", "eres mia","old school",56)
).toDF("name", "hit_songs","ge
浏览 0提问于2020-10-28得票数 1
根据这个，我正在应用udf来过滤CountVectorizer之后的空向量。
val tokenizer = new RegexTokenizer().setPattern("\\|").setInputCol("dataString").setOutputCol("dataStringWords")
val vectorizer = new CountVectorizer().setInputCol("dataStringWords").setOutputCol("features")
val pipeline
浏览 3提问于2018-02-12得票数 1
1回答
我正在尝试一些与pyspark相关的实验，在连接到AWS EMR实例的jupyter笔记本上。我有一个spark dataframe，它从s3读取数据，然后过滤掉一些东西。使用df1.printSchema()输出打印模式，如下所示： root
|-- idvalue: string (nullable = true)
|-- locationaccuracyhorizontal: float (nullable = true)
|-- hour: integer (nullable = true)
|-- day: integer (nullable = true)
|-- da
浏览 39提问于2019-05-09得票数 2
回答已采纳
1回答
我正在使用Spark1.5.0，我有以下问题：
val df = paired_rdd.reduceByKey {
case (val1, val2) => val1 + "|" + val2
}.toDF("user_id","description")
下面是df的示例数据，您可以看到列描述具有以下格式(text1#text3#weight | text1#text3#weight|....)
user1
book1#author1#0.07841217886795074|tool1#desc1#0.27044260397
浏览 4提问于2016-06-28得票数 0
我有地理区域表和地点表，在位置表中我只有geoarea_id，而在地理区域表中我有id和名称 我的目标是创建一个接受geoarea_id数组并在names数组中进行转换的函数 声明这些表： val geoareas =
(
spark.table("location.geoareas")
)
val places =
(
spark.table("location.places")
.select('id, 'name, 'geoareas, 'lat, 'lng)
) 功能： import
浏览 55提问于2019-10-07得票数 0
1回答
请考虑以下Spark2.1代码：
val df = Seq("Raphael").toDF("name")
df.show()
+-------+
| name|
+-------+
|Raphael|
+-------+
val squareUDF = udf((d:Double) => Math.pow(d,2))
df.select(squareUDF($"name")).show
+---------+
|UDF(name)|
+---------+
| null|
+---------+
我为什么要得到
浏览 0提问于2018-02-12得票数 2
回答已采纳
2回答
我有两个数据帧: dataDf和regexDf。dataDf有大量记录，而regexDf有两列正则表达式。我的问题是，我需要根据regexDef中的两列匹配正则表达式的两列来过滤dataDf。我想出了这个
dataDf.registerTempTable("dataTable")
sqlContext.udf.register("matchExpressionCombination", matchExpressionCombination _)
val matchingResults = sqlContext.sql("SELECT * FROM da
浏览 0提问于2015-10-14得票数 1
我使用CountVectorizerModel创建文本外的要素，以便在LDA中进行训练
+-----+--------------------------------------+-------------------------------------------------+-------------------------------------------------------------------+
|label|sentence |words
浏览 0提问于2018-12-17得票数 0
我有一个用Python编写的简单udf，它是我在“24小时内的Apache Spark”一书中的一个代码示例中修改的。这本书使用的是旧版本的Spark，而我运行的是2.3.3。 我确实找到了这个answer，但我很难弄清楚为什么书中的例子不起作用，我也不确定这个答案是否真的解决了我的问题。我在Windows 10上以本地模式运行此程序。 from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
spark = SparkSession \
浏览 26提问于2019-04-05得票数 0
回答已采纳
1回答
我正在一个dataframe中创建一个列，它是一个由4个结构组成的数组。它们中的任何一个都可以是空的，但是由于我需要在这个数组中有固定数量的项，所以我需要在这个事实之后清除空项。但是，当试图使用UDF删除空项时，我会遇到一个错误。下面是一个例子：
创建数据框架，注意其中一个"a“值为None。
spark = SparkSession.builder.getOrCreate()
df = spark.createDataFrame([{"a": "x", "b": "y", "c": "3"
浏览 1提问于2021-05-28得票数 0
回答已采纳
我试图加入两个蜂箱桌，欧米茄和卡，如下：
表omega：
+------+--------+-------+-----+-----+
|pid |enventid|card_id|count|name |
+------+--------+-------+-----+-----+
|111111|"sk" |"pro" |2 |"aaa"|
|222222|"sk" |"pro" |2 |"ddd"|
+------+--------+-------+-----+--
浏览 0提问于2018-04-01得票数 0
回答已采纳
使用python3.6测试现有的代码，但是一些人不知道问题出在哪里，以前使用python2.7的udf是如何工作的。有没有人在本地或分布式环境中遇到类似的问题？类似于
Job aborted due to stage failure: Task 0 in stage 3.0 failed 1 times, most recent failure: Lost task 0.0 in stage 3.0 (TID 202, localhost, executor driver): org.apache.spark.api.python.PythonException: Traceback (mos
浏览 4提问于2020-01-15得票数 1
1回答
给定一个数据帧，其中一列是由以下序列生成的结构序列
val df = spark
.range(10)
.map((i) => (i % 2, util.Random.nextInt(10), util.Random.nextInt(10)))
.toDF("a","b","c")
.groupBy("a")
.agg(collect_list(struct($"b",$"c")).as("my_list"))
df.printSchema
df.show(
浏览 0提问于2017-07-13得票数 12
回答已采纳
有谁有在运行在Windows上的本地pyspark会话上使用的经验吗？我在linux上使用过它们，取得了很好的效果，但是在我的Windows机器上没有成功。
环境：
python==3.7
pyarrow==0.15
pyspark==2.3.4
pandas==0.24
java version "1.8.0_74"
示例脚本：
from pyspark.sql.functions import pandas_udf, PandasUDFType
from pyspark.sql import SparkSession
spark = SparkSession.builde
浏览 0提问于2020-02-19得票数 8
回答已采纳
我正在写一个小的UDF val transform = udf((x: Array[Byte]) => {
val mapper = new ObjectMapper() with ScalaObjectMapper
val stream: InputStream = new ByteArrayInputStream(x);
val obs = new ObjectInputStream(stream)
val stock = mapper.readValue(obs, classOf[util.Hashtable[String, String
浏览 17提问于2020-01-20得票数 0
回答已采纳
我如何使用在spark中工作良好的UDF，比如
sparkSession.sql("select * from chicago where st_contains(st_makeBBOX(0.0, 0.0, 90.0, 90.0), geom)").show
通过spark`s更安全的scala dataframe API获取自？
浏览 4提问于2017-02-22得票数 0
回答已采纳
spark - 2.4.4 sparknlp 2.6.4 python = 3.7.0 transformed_df.show(10) +-----+--------------------+--------------------+
| id| text| finished_lemma|
+-----+--------------------+--------------------+
|73471|Patriots Day Is B...|[Patriots, Day, B...|
|73472|A Break in the Se...|[
浏览 21提问于2020-12-06得票数 0
回答已采纳
我有个火花申请。我的用法是允许用户定义一个类似于Record => Record的任意函数，作为“规则”，它将应用于RDD/Dataset的每个记录。
以下是守则：
//Sample rows with Id, Name, DOB and address
val row1 = "19283,Alan,1989-01-20,445 Mount Eden Road Mount Eden Auckland"
val row2 = "15689,Ben,1989-01-20,445 Mount Eden Road Mount Eden Auckla
浏览 0提问于2019-01-18得票数 1
2回答
创建了一个项目‘spark udf’&编写的单元udf如下所示：
package com.spark.udf
import org.apache.hadoop.hive.ql.exec.UDF
class UpperCase extends UDF with Serializable {
def evaluate(input: String): String = {
input.toUpperCase
}
构建它&为它创建jar。试图在另一个spark程序中使用此udf：
spark.sql("CREATE OR REPLACE FUNCTION up
浏览 4提问于2018-09-04得票数 5
回答已采纳
1回答
我知道有很多这样的答案与我要求的内容有关，但是由于我对scala非常陌生，所以我无法理解这些答案。如果有人能帮我纠正我的UDF，我会非常感激的。
我有这个UDF，它用于完成从GMT到MST的时区转换：
val Gmt2Mst = (dtm_str: String, inFmt: String, outFmt: String) => {
if ("".equals(dtm_str) || dtm_str == null || dtm_str.length() < inFmt.length()) {
null
浏览 17提问于2022-09-20得票数 0
回答已采纳
我正在尝试定义我的udf来清除标签中的html文本。下面的代码运行良好：
from bs4 import BeautifulSoup
from pyspark.sql.functions import udf
text = '<p>Tervetuloa leikkimään, laulamaan, loruilemaan, liikkumaan, taiteilemaan ja tutkimaan leikkipuiston<br>perheaamuun! Leikki- ja toimintaympäristö mahdollistavat vanhemma
浏览 0提问于2020-01-18得票数 0
我在hive中有一个表，其中有两列id(int)和xml_column(string)。xml_column实际上是一个xml_column，但它被存储为字符串。
+------+--------------------+
| id | xml_column |
+------+--------------------+
| 6723 |<?xml version="1....|
| 6741 |<?xml version="1....|
| 6774 |<?xml version="1....|
+------+---------
浏览 6提问于2018-05-31得票数 0
3回答
我正在尝试对PySpark DataFrame中的一些Unicode列进行一些NLP文本清理。我已经尝试了Spark 1.3，1.5和1.6，但似乎无法让它在我的生命中正常工作。我也尝试过使用Python2.7和Python3.4。
我已经创建了一个非常简单的udf，如下所示，它应该只为新列中的每条记录返回一个字符串。其他函数将操作文本，然后在新列中返回更改后的文本。
import pyspark
from pyspark.sql import SQLContext
from pyspark.sql.types import *
from pyspark.sql import SQLConte
浏览 4提问于2016-01-15得票数 21
回答已采纳
当我执行show()时，spark UDF可以工作，但当我对UDF结果执行filter时，它会给我错误。
udf函数
def chkInterPunctuation(sent) :
for char in sent[1:-2] :
if char in ["\"", "'", ".", "!", "?"] :
return True
return False
cip = udf(chkInterPunctuation, BooleanTyp
浏览 0提问于2018-11-21得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
