在Spark DataFrame列中获取不同的单词
，可以通过以下步骤实现：
- 导入必要的库和模块：
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split- 创建SparkSession对象：
spark = SparkSession.builder.appName("WordCount").getOrCreate()- 加载数据源文件为DataFrame：
data = spark.read.text("path_to_file")其中，"path_to_file"是数据源文件的路径。
- 使用split函数将每行文本拆分为单词：
words = data.select(explode(split(data.value, " ")).alias("word"))- 获取不同的单词：
distinct_words = words.distinct()- 打印结果：
distinct_words.show()以上代码将加载文本文件并将每行拆分为单词，然后获取不同的单词并打印出来。
推荐的腾讯云相关产品：腾讯云弹性MapReduce（EMR），它是一种大数据处理和分析的云服务，可以方便地进行Spark等框架的计算任务。详情请参考腾讯云EMR产品介绍：https://cloud.tencent.com/product/emr
相关·内容
1回答
我有一个包含一些单词的CSV文件。总共没有。csv文件中的字数不会超过50k条记录。
我有一个从具有keywords列的JSON文件创建的Spark Dataframe。我需要做的是从dataframe中过滤出其keywords列值与CSV文件中的值匹配的记录。这里，匹配表示csv文件中的单词是否出现在dataframe列中。
举个例子，假设csv文件中有一个单词"baby toys"，spark dataframe看起来像这样
***Keywords***
new baby toys
baby toys for all
costly baby toys price
baby
浏览 2提问于2019-05-31得票数 0
我有一个Spark dataframe，其中包含一个列，其中包含部分行的重复单词： id source_value
1 Peter, Julia, Peter, Michael
2 NULL
3 Michael, Sara, Michael
4 John 我需要创建一个列，其中将排除每行中重复的单词： id result_value
1 Peter, Julia, Michael
2 NULL
3 Michael, Sara
4 John 使用PySpark实现这一点的最佳方法是什么？
浏览 17提问于2021-08-22得票数 0
回答已采纳
1回答
在Spark1.6.0中，我有一个包含职务描述的列的数据框架，如：
Description
bartender
bartender
employee
taxi-driver
...
我使用以下方法从该列检索唯一值列表：
val jobs = people.select("Description").distinct().rdd.map(r => r(0).asInstanceOf[String]).repartition(4)
然后，对于每个职务描述，我尝试检索具有该职务的人员并做一些事情，但我得到了一个NullPointerException：
jobs.foreach
浏览 2提问于2016-02-21得票数 5
回答已采纳
我有一个DataFrame，列text包含一个字符串(或Null)。
如果列text中单词的长度为>= 6和<= 11，那么我想将其与word_list匹配。
如果一个单词匹配，那么这就是新列match的值
import pyspark.sql.functions as F
df = spark.createDataFrame([
["This is line one"],
["This is line two"],
["bla coroner foo bar"],
["This is line three"],
[
浏览 4提问于2021-03-04得票数 1
回答已采纳
我有两个数据帧，第一个数据帧包含实际数据(从CSV文件读取)，第二个数据帧包含一个具有多个关键字的列。即下面的Dataframe-1外观相似(我们必须搜索的地方)：
数据帧-2：
我想要的输出：
我在这里使用spark scala。我想要一个与dataframe-1中的dataframe-2完全匹配的单词。我使用了like、rlike、contains等函数，但它没有给出我想要的输出。有人知道如何在spark scala SQL中或使用dataframe的spark scala函数来开发此逻辑吗？请帮我弄一下这个。
浏览 1提问于2021-02-12得票数 0
我在Pandas中有DataFrame，如下所示：
col1
-------
John One
John Kole Ole
Mike Robe Gut
Michael Spark
如何才能从上面的DataFrame中的列中只显示这些值，该列有两个以上的值，所以或示例只显示John和Mike，因为这些值有两个以上的单词？
如何在Python Pandas中做到这一点？
浏览 3提问于2021-06-25得票数 0
回答已采纳
1回答
如果我试图缓存一个巨大的DataFrame (例如:100 to表)，当我对缓存的DataFrame执行查询时，它会执行完整的表扫描吗？星星之火将如何索引数据。星火文件说：
Spark可以通过调用spark.catalog.cacheTable("tableName")或dataFrame.cache()来使用内存中的列格式来缓存表。然后，Spark将只扫描所需的列，并将自动调整压缩，以尽量减少内存使用和GC压力。您可以调用spark.catalog.uncacheTable("tableName")从内存中删除表。
我没有完全理解上面的语句，如果有人
浏览 8提问于2017-08-14得票数 2
在“火花:最终指南”中，它说：
如果需要引用特定DataFrame的列，则可以在特定的DataFrame上使用col方法。
例如(在Python/Pyspark中)：
df.col("count")
但是，当我在包含列count的dataframe上运行后的代码时，就会得到错误'DataFrame' object has no attribute 'col'。如果我尝试column，我会得到一个类似的错误。
这本书是错的，还是我该怎么做呢？
我上的是星火2.3.1。dataframe是用以下内容创建的：
df = spark.read.f
浏览 2提问于2018-08-12得票数 9
1回答
我试图在Dataframe中替换或更新某些特定的列值，因为我们知道dataframe是不可变的，我试图将其转换为新的dataframe，而不是更新或替换。
我尝试了dataframe.replace，正如在Spark中解释的那样，但是它给了我错误作为错误:值替换不是org.apache.spark.sql.DataFrame的成员
我尝试在option.For下面传递我要传入的数组的多个值
val new_df= df.replace("Stringcolumn", Map(array("11","17","18","10
浏览 0提问于2017-03-22得票数 0
我有一个dataframe，它由两个列组成，全名和姓氏。有时，姓氏列的填写不正确。在这种情况下，在括号之间的全名列中，姓氏将被视为最后一个单词。如果发现括号等于括号之间的单词，我想更新我的姓氏列。
代码
import pandas as pd
df = pd.DataFrame({
'full':['bob john smith','sam alan (james)','zack joe mac', 'alan (gracie) jacob (arnold)'],
'last
浏览 7提问于2022-04-27得票数 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
回答已采纳
我有一个spark数据框，其中的" text“列有一些文本。我想要计算出现各种单词的行数-本质上是出现“术语”的“文档”的数量-以及相关的计数，如最频繁的单词，具有最频繁单词的行(或称为文档)。 我正在使用来自pyspark.ml.feature的HashingTF。但是似乎找不到一种有效的方法来从输出中提取这些信息。 # As an example create a Pandas-df
import pandas as pd
pandas_df = pd.DataFrame({"text": ["The cat jumped over the lazy do
浏览 25提问于2021-08-31得票数 0
回答已采纳
如果我不缓存使用带有limit选项的spark SQL生成的dataframe，当我编辑得到的dataframe并显示它时，我会得到不稳定的结果吗？ 描述。 我有一个类似下面的表格，它是通过使用带有limit选项的spark SQL生成的： +---------+---+---+---+---+
|partition| | 0| 1| 2|
+---------+---+---+---+---+
| 0| 0| 0| 10| 18|
| 1| 0| 0| 10| 17|
| 2| 0| 0| 13| 17|
+---------
浏览 6提问于2021-04-21得票数 0
回答已采纳
我正在尝试计算Spark数据帧中某列中的所有两个可能的单词对之间的相似度。我已经创建了一个UDF和一个数据帧来测试函数，我将它们定义如下： #Similarity Function
def lcs_similarityy(vector):
metric_lcs = MetricLCS()
p = []
for i in vector:
for j in vector:
p.append(1 - metric_lcs.distance(i, j))
return p
#UDF
lcs_similarityyUDF = udf(lambd
浏览 33提问于2021-04-06得票数 0
回答已采纳
1回答
我想把数据库中的org.apache.spark.sql.DataFrame 转换成 org.apache.spark.rdd.RDD[(String, String)] ，有人能帮上忙吗？
背景(更好的解决方案也是受欢迎的)：我有一个Kafka流，它(经过一些步骤)变成了一个2列数据框架。我想把它放到Redis缓存中，第一列作为键，第二列作为值。
更具体地说，输入的类型是：lastContacts: org.apache.spark.sql.DataFrame = [serialNumber: string, lastModified: bigint]。我试着按以下方式对Redis进行分析：
浏览 0提问于2019-03-25得票数 0
回答已采纳
3回答
我有一个Dataframe df，其中包含一个groupID列；也就是说，每个观察都属于一个特定的组。总共有8组。我想从每个groupID中抽取一定百分比的观察结果(例如，20%)。以下是我这样做的方法：
val sample_df = for ( i <- Array.range(0,7) ) yield {
val sel_df = df.filter($"groupID"===i)
sel_df.sample(false,0.2,seed1)
}
该代码的结果是：
Arra
浏览 2提问于2016-07-21得票数 1
回答已采纳
1回答
我正试图用字典来替换dataframe句子中的单词。我怎样才能取代原来的数据？
字典: rep_vocab包含{wrong words: correct words} dataframe: data_test列‘质询1’句子列‘d5’包含一组单词，这些单词在句子中拼写错误。我用这个列快速定位出有错误单词的句子行。
我的代码：
data_test.loc[data_test['d5']!=set()['question1'].replace(rep_vocab,regex=True)
它返回正确的结果，但是dataframe中的原始值不会改变。我尝试过其他方式，比
浏览 0提问于2019-08-20得票数 0
2回答
在与列表匹配的列中保持单词的火花
、、、、
我现在有一个列表和一个Spark数据文件：
['murder', 'violence', 'flashback', 'romantic', 'cult', 'revenge', 'psychedelic', 'comedy', 'suspenseful', 'good versus evil']
我很难找到在dataframe中创建一个新列的方法，它从标记列中为每一行获取第一个匹配单词，并将其放入新创建的该行列中。
例如，假设
浏览 11提问于2022-05-29得票数 0
回答已采纳
我在spark中有一个dataFrame列的收集数据
temp = df.select('item_code').collect()
Result:
[Row(item_code=u'I0938'),
Row(item_code=u'I0009'),
Row(item_code=u'I0010'),
Row(item_code=u'I0010'),
Row(item_code=u'C0723'),
Row(item_code=u'I1097'),
Row(item_
浏览 0提问于2017-10-02得票数 0
1回答
阅读<列名>栏的内容
、、、、
我用的是火花1.5.0
我创建了如下所示的数据框架，并试图从这里读取一列
>>> words = tokenizer.transform(sentenceData)
>>> words
DataFrame[label: bigint, sentence: string, words: array<string>]
>>> words['words']
Column<words>
我想读句子中的所有单词(单词)。我怎么看这个？
编辑1：仍然存在错误
现在，我在Spark2.0.0中运行了这个程序，并得到
浏览 2提问于2016-12-22得票数 0
回答已采纳
1回答
我有一个有多列的spark DataFrame。我想根据一列对行进行分组，然后为每组找到第二列的模式。与熊猫DataFrame一起工作时，我会这样做：
rand_values = np.random.randint(max_value,
size=num_values).reshape((num_values/2, 2))
rand_values = pd.DataFrame(rand_values, columns=['x', 'y'])
rand_values['x'] = ra
浏览 1提问于2016-04-16得票数 10
回答已采纳
我在熊猫身上有一种情况。我有一个excel文件，它有一个名为item的列，它有一些文本。我有另一个dataframe有一个名为brand的列。如果项目列文本中存在品牌字符串，我希望在项目列前面添加品牌名称。
这是我的excel文件快照。
我的品牌数据图如下所示。
brand_df = pd.DataFrame({'brand':['spark','hadoop','hive']})
我想看看brand_df品牌是否存在于item_df中。如果品牌存在，那么它应该以update_column的名字出现在项目前面，如下所示
浏览 0提问于2019-07-09得票数 0
回答已采纳
1回答
我已经使用S3数据格式在pyspark.pandas中读取和存储了拼花文件。现在，在第二阶段，我试图在databricks中读取pyspark dataframe中的parquet文件，并面临将嵌套的json列转换为适当列的问题。首先，我使用以下命令从S3读取拼图数据：
adf = spark.read.parquet('s3://path')
我的pyspark dataframe中的一个嵌套列如下所示：
event_params:array
element:struct
key:string
value:struct
dou
浏览 8提问于2022-06-07得票数 0
问题陈述:在升级Databricks运行时版本时，复制列在创建dataframe时抛出错误。在较低的运行时，会创建dataframe，并且由于下游不需要重复列，因此它只是在select中被排除在外。
文件位置:存储在ADLS Gen2 (Azure)上的Json文件。集群模式:标准
代码:我们在中阅读它，如下所示。
intermediate_df = spark.read.option("multiline","true").json(f"{path}/IN-109418_Part_1.json")
json文件是嵌套的，其中一个是tags，它是
浏览 1提问于2021-11-16得票数 2
回答已采纳
1回答
我有一个由文本列组成的dataframe。有些单词也有数字，后面跟着单词。我想把数字和单词分开，在它们之间加一个空格。
例如：
Machine1234 -> Machine 1234
5years -> 5 years
下面是我的数据
+---+--------------------------------------------+
|id |words |
+---+--------------------------------------------+
|0 |This is Spark123 o
浏览 7提问于2022-07-25得票数 2
回答已采纳
我希望将spark dataframe列的值从小时分钟秒转换为
例如"01:12:17.8370000“
将变成4337，谢谢你的评论。
或者"00:00:39.0390000“
会变成39秒。
我已经读过这个问题，但是我不知道如何使用这个代码来转换我的spark dataframe列。
像这样的东西
df.withColumn("duration",col("duration")....)
我使用的是scala 2.10.5和spark 1.6
谢谢
浏览 12提问于2017-07-31得票数 2
回答已采纳
我已经创建了一个空的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
回答已采纳
我正在使用一个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
回答已采纳
如何有效地在 DataFrame 中创建一个新列，该列是 spark 中其他行的函数
这是我描述的问题的spark实现
from nltk.metrics.distance import edit_distance as edit_dist
from pyspark.sql.functions import col, udf
from pyspark.sql.types import IntegerType
d = {
'id': [1, 2, 3, 4, 5, 6],
'word': ['cat', 'hat'
浏览 0提问于2018-01-09得票数 0
回答已采纳
我在dataframe列中有一个s3路径，我需要该路径的目录大小。在spark scala中有没有什么方法可以让我们在一个新的列中获得s3目录的大小。
浏览 24提问于2021-05-10得票数 0
回答已采纳
3回答
我遵循Spark文档中的来计算一系列文档的TF-IDF。Spark使用散列技巧进行此计算，因此在最后你会得到一个包含散列单词和相应权重的Vector，但是...我怎样才能从散列中取回单词？
我真的需要对所有的单词进行散列，并将它们保存在映射中，以便稍后迭代查找关键字吗？有没有更有效的方式内置Spark？
提前感谢
浏览 0提问于2014-11-10得票数 5
3回答
我使用的是Spark 1.3.1。
我正在尝试查看Python中Spark dataframe列的值。有了Spark dataframe，我可以使用df.collect()来查看数据帧的内容，但在我看来，Spark dataframe列还没有这样的方法。
例如，数据帧df包含一个名为'zip_code'的列。所以我可以做df['zip_code']，它会变成一个pyspark.sql.dataframe.Column类型，但是我找不到一种方法来查看df['zip_code']中的值。
浏览 1提问于2015-06-30得票数 43
回答已采纳
我有一个有bigint列的Dataframe。如何将bigint列转换为scala spark中的时间戳
浏览 1提问于2019-07-23得票数 4
回答已采纳
1回答
使用Spark标记文本内容？
、、、、
我致力于实现一项要求，即使用apache和mongodb为文档创建一个单词字典。
在我的场景中，我有一个mongo集合，其中每个文档都有一些文本类型字段以及一个文档所有者的字段。
我希望解析集合文档中的文本内容，并创建一个字典，将单词映射到文档和所有者字段。基本上，键将是一个word，值将是_id和owner字段。
这样做的目的是根据用户的文档在用户界面中输入文本框时提供特定于用户的自动建议。用户可以创建多个文档，一个单词可以包含在多个文档中，但只有一个用户能够创建文档。
我使用了mongo火花连接器，并且能够使用spark将集合文档加载到数据框架中。
我不知道如何处理文本数据，这是在其中一个
浏览 0提问于2017-05-07得票数 0
1回答
因此，我正在尝试将python算法转换为Spark友好代码，并且遇到了以下问题：
indexer = recordlinkage.SortedNeighbourhoodIndex \
(left_on=column1, right_on=column2, window=41)
pairs = indexer.index(df_1,df_2)
它基本上比较一列和另一列，并为可能相同的列生成索引对(记录匹配)。
我的代码：
df1 = spark.read.load(*.csv)
df2 = spark.read.load(*.csv)
func_udf =
浏览 0提问于2018-07-25得票数 0
回答已采纳
批处理之后，我需要向Kafka主题写入包含多个不同列的结果DataFrame。
根据下面的火花文档，写入Kafka的Dataframe在模式中应该有以下强制列：
值(必需)字符串或二进制
正如我前面提到的，我有更多的列有值，所以我有一个问题--如何正确地将整个DataFrame行作为一条消息从我的Spark应用程序发送到Kafka主题？我是否需要用一个值列(包含联接值)将所有列的所有值连接到新的DataFrame中，或者有更正确的方法来实现它？
浏览 0提问于2018-11-23得票数 0
回答已采纳
2回答
我有一个Spark (scala) dataframe，其中的一些dataframe列是可选的，也就是说，有时它们并不存在。有没有一种非常简单的方法来修改我的df.select语句，使spark不关心列可能不存在？
例如，现在我有：df.select(Seq(col("col1"), col("optionalCol"), col("col2")))。
我希望会有某种“可选”的称谓。
浏览 1提问于2018-07-03得票数 4
1回答
我正在dataframe1中读取CSV文件，然后在dataframe2中筛选一些列，在从dataframe1中选择dataframe2列时，我想将我的函数应用到列值上。喜欢
import utilities._
val Logs = sqlContext.read
.format("csv")
.option("header", "true")
.load("dbfs:/mnt/records/Logs/2016.07.17/2016.07.17.{*}.csv")
val Log = Logs.select(
&
浏览 0提问于2018-11-04得票数 0
回答已采纳
我正在尝试对org.apache.spark.sql.DataFrame类型的Dataframe中的列列表进行汇总，并创建一个新的列‘sum’和dataframe 'out‘。
如果我手动列出列，我可以很容易地做到这一点，例如，这是可行的。
val columnsToSum = List(col("led zeppelin"), col("lenny kravitz"), col("leona lewis"), col("lily allen"))
val out = df3.withColumn("sums
浏览 2提问于2020-11-25得票数 2
回答已采纳
2回答
我有一张桌子 DEST_COUNTRY_NAME ORIGIN_COUNTRY_NAME count
United States Romania 15
United States Croatia 1
United States Ireland 344 我把上面的代码转换成了DataFrame val flightData2015 = spark
.read
.option("inferSchema", "true")//infers the input schema automatically from data
.option("he
浏览 19提问于2019-02-08得票数 0
回答已采纳
对于推荐系统，我需要计算整个Spark DataFrame的所有列之间的余弦相似度。
在Pandas中，我经常这样做：
import sklearn.metrics as metrics
import pandas as pd
df= pd.DataFrame(...some dataframe over here :D ...)
metrics.pairwise.cosine_similarity(df.T,df.T)
这会生成列之间的相似性矩阵(因为我使用了转置)
有没有办法在Spark (Python)中做同样的事情？
(我需要将它应用于由数千万行和数千列组成的矩阵，所以这就是为什么我
浏览 3提问于2017-05-12得票数 13
我试图使用Pyspark在dataframe中更改一个列，也就是跨行对单个列的所有值进行洗牌。
我试图避免这样的解决方案，即在将列拆分并分配索引列之前，将其重新连接到原始的dataframe，而原始dataframe也有一个添加的索引列。主要是因为我的理解(这可能是非常错误的)，在大型数据集(数百万行)的运行时，联接是不好的。
# for some dataframe spark_df
new_df = spark_df.select(colname).sort(colname)
new_df.show() # column values sorted nicely
spark_df.with
浏览 0提问于2019-06-06得票数 0
我有一个Spark流，类似于：
dataFrame
.join(anotherDataFrame, columns)
.repartition(partitionColumn)
.save()
在join步骤中，我希望spark.sql.adaptive.enabled是true，因为这将加快连接的速度。
在repartition步骤中，我希望spark.sql.adaptive.enabled是false，因为如果是真的，它可能会更改分区，保存的结果将被格式化为错误的分区。
如何在Spark流的不同步骤中更改spark.sql.adaptive.enabled的行为？
例如:当只
浏览 0提问于2019-01-28得票数 2
1回答
我正在使用app.zelp.com来执行NLP。在标记化并删除停用词之后，我想要取消标记化剩余的单词并导出到csv。这有可能吗？
%python
# Start Spark session
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("StopWords").getOrCreate()
from pyspark.ml.feature import Tokenizer, StopWordsRemover
from pyspark import SparkFiles
url =
浏览 18提问于2021-02-17得票数 1
我正在使用下面的语句在spark中读取csv。
df = spark.read.csv('<CSV FILE>', header=True, inferSchema = True)
当我检入spark dataframe时，一些整型和双精度列被存储为dataframe中的字符串列。但是，并不是所有的列都是这样。
我已经检查了特定列的值，所有的值都是双精度类型，但spark仍然推断为StringType。
因为我加载的CSV文件大约有1000列，所以也显式地指定模式是不可行的。
如有任何建议或帮助，我们将不胜感激。
致以敬意，
Neeraj
浏览 5提问于2017-08-31得票数 2
我有两个DataFrame数据帧(A和B)，它们都有一个公共的列/字段(在Spark A中是主键，但在B中不是)。
对于dataframe A中的每一条记录/行，dataframe B中有多条记录。基于该公共列值，我希望针对dataframe A中的每条记录从dataframe B中获取所有记录。
我应该执行什么类型的转换，以便在不做太多混洗的情况下将记录收集在一起？
浏览 0提问于2018-07-25得票数 0
我已经使用java spark dataframe将CSV读取到dataframe中，现在我必须对每个列应用一些宽度，并将数据写入固定宽度的文件中。因为example..column 1有2位宽，列2有7个bit...like，而我有85列。谁能解释一下如何使用java spark将dataframe中的数据写入到固定宽度的文件中？我只需要java spark中的解决方案
浏览 0提问于2020-11-19得票数 0
2回答
我有一个包含两个数组列的Spark dataframe：
+------------------------------------------------------+-----------------+
| var1| var2|
+------------------------------------------------------+-----------------+
| [black tea, green tea, tea, yerba mate
浏览 3提问于2022-08-01得票数 2
回答已采纳
下面的语句满足了我的需要，但它只能在spark-shell中运行，而不能在scala程序中运行。
spark.read.json(dataframe.select("col_name").as[String]).schema
我将dataframe转换为rdd并通过，它工作得很好(我遵循下面的链接)，但它只有在以下情况下才能工作。
我只有json列值，当我传递其他cols (直接col值)时，它无法提供
输出。
我有一个解决方案，可以在spark-shell下运行，但不能在scala程序中运行。
浏览 14提问于2020-03-12得票数 0
3回答
我有一个蜂巢表如下：
hive> describe stock_quote;
OK
tickerid string
tradeday string
tradetime string
openprice string
浏览 0提问于2018-04-25得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
