在Pyspark dataframe中拆分输入日志文件
在Pyspark中，可以使用DataFrame API来拆分输入日志文件。DataFrame是一种分布式数据集合，可以进行高效的数据处理和分析。
拆分输入日志文件的步骤如下：
- 导入必要的库和模块：
from pyspark.sql import SparkSession
from pyspark.sql.functions import split- 创建SparkSession对象：
spark = SparkSession.builder.appName("LogSplit").getOrCreate()- 读取输入日志文件并创建DataFrame：
log_df = spark.read.text("input.log")- 使用split函数拆分日志行：
split_df = log_df.select(split(log_df.value, " ").alias("log_data"))- 展开拆分后的数据：
expanded_df = split_df.selectExpr("log_data[0] as column1", "log_data[1] as column2", ...)这里的column1、column2等是根据日志文件中的字段进行命名的。
- 可选：对数据进行进一步处理和转换，例如数据类型转换、过滤等。
- 可选：将处理后的数据保存到文件或数据库中。
下面是一个示例代码，演示如何在Pyspark DataFrame中拆分输入日志文件：
from pyspark.sql import SparkSession
from pyspark.sql.functions import split
# 创建SparkSession对象
spark = SparkSession.builder.appName("LogSplit").getOrCreate()
# 读取输入日志文件并创建DataFrame
log_df = spark.read.text("input.log")
# 使用split函数拆分日志行
split_df = log_df.select(split(log_df.value, " ").alias("log_data"))
# 展开拆分后的数据
expanded_df = split_df.selectExpr("log_data[0] as column1", "log_data[1] as column2")
# 显示拆分后的数据
expanded_df.show()
# 停止SparkSession
spark.stop()这个示例代码中，我们假设输入日志文件的每一行由空格分隔的两个字段组成。使用split函数将每一行拆分成一个数组，然后使用selectExpr函数将数组中的元素展开为列。最后，使用show函数显示拆分后的数据。
对于Pyspark的DataFrame API的详细介绍和更多用法，可以参考腾讯云的产品文档：Pyspark DataFrame API。
相关·内容
我正在使用Pyspark dataframe来处理日志文件。下面是我使用spark.read.text从文件中读取的示例日志 [Wed Oct 11 14:32:52 2000] [error] [client 127.0.0.1] client denied by server configuration: /export/home/htdocs/test
[Wed Oct 11 14:32:52 2000] [error] [client 127.0.0.1] client denied by server configuration: /export/home/htdocs/test
浏览 12提问于2020-07-08得票数 0
回答已采纳
1回答
我有大约500-700 csv文件的日志加载每周。文件名是每周随机生成的。这些文件存储在google云存储上，pyspark在google dataproc上运行，有几个工作节点。
我已经创建了一个pyspark文件，它基本上只是迭代文件并按原样加载到bigquery中。我以为我会在bigquery表上看到增量加载，但似乎pyspark会将整个gz输入文件合并到dataframe中，然后立即处理它。所以我在大约90分钟后得到了结果。
我如何编写pyspark代码，以便每次读取文件时，都会将其写入bigquery。总体时间可能不会有太大变化，但至少我如何才能看到部分写入bigquery的数据？
浏览 22提问于2021-10-10得票数 1
1回答
我想根据DataFrame中一行的第一个字符来分割文件。
原始数据有一列，数据包括
文件名(如'DATE20191009') 文件内容(如'1‘、'2’、'3')
输入样例文件(Pyspark)：
column1
Date20191009
1
2
3
Date20191010
1
4
5
我想要一个DataFrame文件名作为数据的分割。
文件名放在DataFrame的column2中，文件的内容放置在DataFrame的column2中。
预期输出(Pyspark)
column1 column2
Date20191009 [
浏览 3提问于2019-10-10得票数 1
回答已采纳
我有一个CSV文件，它包含JSON对象以及其他数据，比如String，Integer。如果我尝试将文件读取为CSV，那么JSON对象将在其他列中重叠。
Column1, Column2, Column3, Column4, Column5
100,ABC,{"abc": [{"xyz": 0, "mno": "h"}, {"apple": 0, "hello": 1, "temp": "cnot"}]},foo, pine
101,XYZ,{"xyz&#
浏览 2提问于2020-09-23得票数 1
我试图使用pyspark.pandas API在数据库上构造数据的共生矩阵。
我试着用这个方法构造矩阵。
该代码在熊猫中运行良好，但在使用pyspark.pandas时出现了错误。
coocc = psdf.T.dot(psdf)
coocc
我得到了这个错误
TypeError: Unsupported type DataFrame
我查过医生了。
pyspark.pandas.DataFrame.dot()
以串联作为输入。
我尝试使用psdf.squeeze()将dataframe转换为系列化，但它并不将dataframe转换为串联，因为我的dataframe有多个列。
有没有办法将py
浏览 8提问于2022-10-14得票数 0
回答已采纳
1回答
我是PySpark新手，正在尝试读取HDFS文件(上面创建了hive表)并创建PySpark数据帧。通过PySpark读取配置单元表非常耗时。有没有什么方法可以动态获取hive列名(用作dataframe中的模式)？ 我希望将文件位置、表名和数据库名作为输入传递给aa程序/函数，以便从配置单元元数据(可能是元数据xml)中获取模式/列名称，并作为dataframe返回。 请指教
浏览 15提问于2019-03-06得票数 0
回答已采纳
1回答
在PySpark中，如何将Dataframe转换为普通字符串？
背景：
我将PySpark与Kafka一起使用，而不是硬编码代理名称，而是在PySpark中参数化了Kafka broker名称。
Json文件保存了代理详细信息，Spark读取这个Json输入并将值赋值给变量。这些变量为带有字符串的Dataframe类型。
当我将dataframe传递给Pyspark连接细节以替换值时，我将面临问题。
错误：
只能将字符串(不是Dataframe)连接到字符串。
Json参数文件：
{
"broker": "https://at.com:8082",
"t
浏览 0提问于2021-03-05得票数 0
我有一个按日期分区的S3存储桶列表。第一个存储桶标题为2019-12-1，第二个存储桶标题为2019-12-2，依此类推。 这些存储桶中的每一个都存储我正在读取到pyspark dataframe中的拼图文件。从每个存储桶生成的pyspark dataframe具有完全相同的模式。我想要做的是迭代这些存储桶，并将所有这些拼图文件存储到一个单独的pyspark dataframe中，该数据框有一个date列，表示dataframe中的每个条目实际来自哪个存储桶。 因为单独导入每个存储桶时生成的数据帧的模式是多层的(即每一行都包含结构数组的结构等)，所以我认为将所有存储桶合并到一个数据帧中的唯一
浏览 13提问于2019-12-16得票数 0
回答已采纳
我对pyspark和json解析还是个新手，我被困在了某些特定的场景中。让我先解释一下我要做什么，我有一个json文件，其中有一个数据元素，这个数据元素是一个包含另外两个json对象的数组。给定的json文件如下所示
{
"id": "da20d14c.92ba6",
"type": "Data Transformation Node",
"name": "",
"topic": "",
"x": 380,
浏览 142提问于2019-06-05得票数 0
回答已采纳
1回答
我有一个在本地运行的PySpark脚本，spark-submit在Docker中运行。在我的脚本中，我在PySpark DataFrame上有一个调用DataFrame，然后我对DataFrame进行了各种操作，最后调用to_csv()将结果写到一个本地CSV文件中。
当我运行这个脚本时，调用toPandas()之后的代码似乎不会运行。在这个方法调用之前和之后，我都有日志语句，但是只有调用之前的日志条目显示在spark-submit控制台输出中。我曾经想过，这可能是因为其余的代码是在一个单独的executor进程中运行的，所以日志不会显示在控制台上。如果这是真的，我如何才能看到我的应用程序日志
浏览 4提问于2022-09-27得票数 0
我正在研究火花源。所以，下面是我如何构建环境的： 1. ubuntu in virtual machine
2. downloading spark 2.4.0
3. install pyspark using pip
4. configuring environment path:
export SPARK_HOME="/home/feng/Downloads/spark-2.4.0-bin-hadoop2.7/"
export PATH=$SPARK_HOME/bin:$PATH
export PYSPARK_DRIVER_PYTHON=jupy
浏览 23提问于2019-02-08得票数 0
回答已采纳
1回答
我有一个5节点集群，我正在使用pyspark将一个100k的csv文件加载到一个dataframe中，并执行一些etl操作，并将输出写入到一个parquet文件中。当我加载数据帧时，如何在每个执行器处理20k记录的所有执行器0上均匀地划分数据集。
浏览 2提问于2017-10-06得票数 1
我有许多结构混乱的JSON文件。我想把这些解析成一个PySpark DataFrame。我编写了一个解析函数，并希望将其应用于目录中的每个文件，并将其输入到我的DataFrame中。
我对PySpark完全陌生，所以任何帮助都是非常感谢的。我曾希望我能做一些事情来达到这样的效果：
from my_parsers import parser
data_files = [...] # Is there a PySpark way to iterate through files in a directory?
my_data = map(parser, data_files) # How do
浏览 4提问于2022-02-28得票数 0
回答已采纳
我用火花来处理我的数据，就像这样：
dataframe_mysql = spark.read.format('jdbc').options(
url='jdbc:mysql://xxxxxxx',
driver='com.mysql.cj.jdbc.Driver',
dbtable='(select * from test_table where id > 100) t',
user='xxxxxx'
浏览 2提问于2022-05-11得票数 0
回答已采纳
如果在PySpark中将任何类型的登录添加到UDF函数中，它将不会出现在任何地方。这是一种实现这一目标的方法吗？
到目前为止，我尝试了标准的python日志记录、py4j和打印。
我们在AWS EMR集群上运行带有YARN集群管理器的PySpark 2.3.2。
例如。下面是我想使用的一个函数：
def parse_data(attr):
try:
# execute something
except Exception as e:
logger.error(e)
return None
我将其转换为UDF：
import pys
浏览 0提问于2019-10-15得票数 5
4回答
我有一个具有下面数据(所有列都有字符串数据类型)的Pyspark (原始Dataframe)。在我的用例中，我不确定输入数据中有哪些列。用户只需将dataframe的名称传递给我，并要求我修剪这个dataframe的所有列。典型数据文件中的数据如下所示：
id Value Value1
1 "Text " "Avb"
2 1504 " Test"
3 1
浏览 2提问于2017-07-31得票数 0
1回答
我是PySpark的新手，我想读取一个日志文件，其中包含很多行二进制代码，用换行符隔开。我需要使用以下方法过滤该文件：
\x00二进制行的长度大于1二进制行的长度以开头
下面是一个输入文件中的一个示例行：
b'\x18\xb5\x1fM\x00\x02\x00\x^C\x05\x00\x00\x96\x93\x80@2\xf6\x1f2\x01\n'
在检查\x00每行的0位置时，我遇到了一个错误。错误是：
pyspark.sql.utils.AnalysisException:无法从b#2中提取值:需要结构类型，但得到字符串；
这是我的密码。
from pyspark i
浏览 0提问于2021-02-03得票数 0
2回答
我在java中有一个类，它构建了一些复杂的星火DataFrame。
package companyX;
class DFBuilder {
public DataFrame build() {
...
return dataframe;
}
}
我将这个类添加到pyspark/木星类路径中，这样它就可以被py4j调用。现在，当我称之为它时，我得到了奇怪的类型：
b = sc._jvm.companyX.DFBuilder()
print(type(b.build()))
#prints: py4j.java_gateway.JavaObject
VS
浏览 0提问于2018-03-14得票数 1
回答已采纳
1回答
问题
最近，我在中遇到了一个挑战，当时我试图读取一个大型UTF-8JSONArray文件，并切换到HDInsight PySpark (v2.x，而不是3)来处理该文件。该文件为~110 g，具有~150 Objects对象。
HDInsight PySpark似乎不支持JSON文件格式的数组输入，所以我被困住了。而且，我有“许多”这样的文件，每个文件中都包含不同的模式，每个列都包含100列，因此现在不能为这些列创建模式。
问题
如何在PySpark 2中使用HDInsight上的开箱即用功能来使这些文件被读取为JSON？
谢谢,
J
Things我试过
我使用了这个页面底部的方法：，它提供了
浏览 1提问于2018-02-10得票数 0
回答已采纳
所以我有一个在线托管的.gz日志文件，如下所示
https://example.com/sample.log.gz
我可以使用下面的代码将其加载到Python列表中：
import urllib2
from StringIO import StringIO
import gzip
request = urllib2.Request('https://example.com/sample.log.gz')
response = urllib2.urlopen(request)
buf = StringIO(response.read())
f = gzip.GzipFile(f
浏览 3提问于2017-01-30得票数 0
2回答
在PySpark中分解
、、、
我想将包含单词列表的DataFrame转换为DataFrame，每个单词都在自己的行中。
如何在DataFrame中的列上分解？
下面是我尝试过的一个示例，您可以取消注释每一行代码，并获得以下注释中列出的错误。我在Python2.7和Spark 1.6.1中使用PySpark。
from pyspark.sql.functions import split, explode
DF = sqlContext.createDataFrame([('cat \n\n elephant rat \n rat cat', )], ['word'])
print '
浏览 2提问于2016-07-06得票数 30
回答已采纳
2回答
我有一个pyspark dataframe，其中一些列具有相同的名称。我想将具有相同名称的所有列合并到一列中。例如，输入dataframe： ? 我如何在pyspark中做到这一点？任何帮助都将不胜感激。
浏览 57提问于2021-06-18得票数 2
回答已采纳
1回答
我正试图使用Databricks DataFrame包和，从EMR集群上的S3上的CSV源创建一个
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
df = sqlContext.read.format('com.databricks.spark.csv').options(header='true').load('s3n://h2o-airlines-unpacked/allyears.csv')
df.first()
这不会在4m3.xlarge的集群上终止。我
浏览 0提问于2015-08-28得票数 3
我有一个非常大的每日gzipped文件序列。我正在尝试使用PySpark重新保存所有S3格式的文件，以供以后使用。
如果对于单个文件(例如，2012-06-01)，我这样做：
dataframe = spark.read.csv('s3://mybucket/input/20120601.gz', schema=my_schema, header=True)
dataframe.write.parquet('s3://mybucket/output/20120601')
它可以工作，但是因为gzip是不可拆分的，所以它运行在单个主机上，并且我没有从使用集群中获得
浏览 2提问于2017-12-16得票数 0
我有一个带有一条记录的输入csv文件。当我在pyspark中读取文件时，dataframe分别有三列a、b、c。A和c有数据，b有4个空格的数据。将文件写入csv时，4个空格的数据丢失，它将以空字符串的形式写入文件。
输入文件：
aaaa, , bbbb
输出文件：
aaaa,"", bbbb
如何按原样保留这4个空间的数据？
浏览 19提问于2020-02-20得票数 0
5回答
我正在开发一个包含n列的PySpark DataFrame。我有一组m列(m < n)，我的任务是选择其中包含最大值的列。
例如：
输入: PySpark DataFrame，包含：
col_1 = [1,2,3], col_2 = [2,1,4], col_3 = [3,2,5]
输出：
col_4 = max(col1, col_2, col_3) = [3,2,5]
正如问题中所解释的那样，熊猫身上也有类似的东西。
在PySpark中是否存在这样的方法，或者我是否应该将PySpark df转换为Pandas，然后执行这些操作？
浏览 7提问于2016-06-07得票数 29
回答已采纳
我有一个用例来映射基于条件的pyspark列的元素。通过这个文档，我找不到一个函数来执行映射函数。
因此，尝试使用pyspark映射函数，但无法将pyspark列转换为dataFrame
注意:我之所以使用pyspark列，是因为我从我使用的库(远大期望)中获得了它的输入。
@column_condition_partial(engine=SparkDFExecutionEngine)
def _spark(cls, column, ts_formats, **kwargs):
return column.isin([3])
# need to replace the abov
浏览 4提问于2021-11-19得票数 0
回答已采纳
1回答
读取几行列数不同的CSV文件
、、、、
我在ADLS 2中有一个CSV文件，在几行中有不同的列计数。第一排有8列，第二排有12列。当我试图用它来创建Pyspark dataframe时，它创建了基于第一行的df，即带有8列的dataframe，而忽略了第二行的4列。
我也尝试过使用ADF映射数据流，但是正在创建相同的模式。
我还试图生成模式，因为我知道最大列是什么以及它们的名称，但问题是，它将忽略具有8列的行，并且只会在生成的模式中有相同数目的记录，即12。
我注意到它基于第一行文件创建了头。
由于文件来自客户端，所以列最少的数据可以出现在第一行或中间或最后的某个位置。
我注意到它基于第一行文件的创建头
如何实现共有12列，第一行最
浏览 1提问于2022-06-30得票数 -1
我有一个类似下面的pyspark脚本。在这个脚本中，我遍历表名的input文件并执行代码。
现在，我想在每次迭代函数mysql_spark时分别收集日志。
例如：
input file
table1
table2
table3
现在，当我执行pyspark脚本时，我将所有三个表的日志保存在一个文件中。
What I want is 3 separate log files 1 for each table
Pyspark脚本：
#!/usr/bin/env python
import sys
from pyspark import SparkContext, SparkConf
from py
浏览 1提问于2017-07-29得票数 0
回答已采纳
我已经在pyspark中创建了一个spark dataframe，我想要将过滤后的输出数据写入日志文件或文本文件。让我们把下面的代码看作df df=spark.sql(select * from tbl1) 我希望以下命令的输出为文本文件或日志文件 df.select("col1").show(5) 我试过下面的方法，但不起作用 with open("output.txt","a") as f:
f.write(tabulate(df.select("col1").show(5),tablefmt="prett
浏览 27提问于2020-04-16得票数 0
1回答
在解析pyspark中不一致的数据类型时，我遇到了一个问题。如下面的示例文件所示，SA键总是包含一个字典，但有时它可以显示为string值。当我尝试获取列SA.SM.Name时，会得到如下所示的异常。
对于SA.SM.Name列，如何将null设置为除了JSONs之外的值的pyspark/hive。有人能帮帮我吗？
我试图转换成不同的数据类型，但是没有工作，或者我可能做错了什么。
输入文件内容: mypath
{"id":1，"SA":{"SM"：{"Name"："John"，"Email"：“Jo
浏览 6提问于2020-10-10得票数 0
1回答
多列上的多聚合
、、
我使用Python在Pyspark框架中。我试图使用groupby在不同的列上应用不同的聚合。
我有一个包含col1、col2、col3、col4列的df，我想做这样的事情：df.groupby("col1").sum("col2", "col3").avg("col4")
但是我发现了一个错误：
"/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py"，'DataFrame‘对象没有属性'avg’回溯(最近一次调用)：文件第1
浏览 4提问于2019-09-27得票数 0
回答已采纳
我只有一列的dataframe。我想使用pyspark中的pandas_udf拆分字符串。因此，我有以下代码： from pyspark.sql.functions import pandas_udf, PandasUDFType
@pandas_udf('str')
def split_msg(string):
msg_ = string.split(" ")
return msg_
temp = temp.select("_c6").withColumn("decoded",
split_msg(temp._c
浏览 17提问于2019-10-03得票数 0
回答已采纳
1回答
我正在使用HDInsight spark集群来运行我的Pyspark代码。我尝试从postgres表中读取数据，并将其写入如下所示的文件中。pgsql_df返回的是DataFrameReader而不是DataFrame。所以我无法将DataFrame写到文件中。为什么"spark.read“返回DataFrameReader。这里我漏掉了什么？ from pyspark.sql.types import *
from pyspark.sql import SparkSession
from pyspark import SQLContext
from pyspark import S
浏览 40提问于2020-04-24得票数 0
回答已采纳
我正在尝试测试一个实用函数，它接受一个spark DataFrame，并在经过一些转换后输出一个新的spark DataFrame。在尝试模拟测试数据时，我需要为输入构造一个pyspark dataframe。我看到的大多数使用这种用法的例子 spark.createDataFrame(data, columns) 我对文档不太熟悉，找不到"spark“。您如何使用from pyspark* import spark
浏览 8提问于2020-08-12得票数 0
回答已采纳
所以我想从一个目录中读取csv文件，作为pyspark dataframe，然后将它们附加到单个dataframe中。而不是像我们在熊猫身上做的那样，在pyspark中得到替代方案。
例如，在熊猫中，我们这样做：
files=glob.glob(path +'*.csv')
df=pd.DataFrame()
for f in files:
dff=pd.read_csv(f,delimiter=',')
df.append(dff)
在Pyspark中，我已经尝试过了，但没有成功
schema=StructType([])
union_d
浏览 5提问于2017-04-10得票数 11
从API生成的日志是无效的JSON格式。就像这样。
{"name": Power Amp,Component: Power Amplifier\n1/2 Inductor \n3 Power Capacitor\n Semiconductor\n Software\TV\n wafer\n Sony Dolby, \n \n L, pin, Or Amp"}
{name: Signal , Component: "1 make\n1 model, Halved \n1-1/2nd Rectifier\n Diode, to \n5 microwave\n8 he
浏览 2提问于2020-01-19得票数 0
1回答
将一列拆分为多列的问题
、、、、
在PySpark中，我很难将dataframe的列拆分为更多的列：
我有一个列表列表，我想将其转换为一个dataframe，在一个列中的每个值。
我试过的是：
我从这个列表中创建了一个数据文件：
[['COL-4560', 'COL-9655', 'NWG-0610', 'D81-3754'],
['DLL-7760', 'NAT-9885', 'PED-0550', 'MAR-0004', 'LLL-5554']]
使用此代码：
from pysp
浏览 2提问于2020-08-19得票数 1
回答已采纳
我正试图在我的数据集上运行PySpark中的PySpark算法。
from pyspark.ml.fpm import FPGrowth
fpGrowth = FPGrowth(itemsCol="name", minSupport=0.5,minConfidence=0.6)
model = fpGrowth.fit(df)
我得到了以下错误：
An error occurred while calling o2139.fit.
: java.lang.IllegalArgumentException: requirement failed: The input
col
浏览 0提问于2018-04-05得票数 2
回答已采纳
我正在尝试废弃一个XML文件，并从XML文件上的标签创建一个数据帧。我使用pyspark开发Databricks。
XML文件：
<?xml version="1.0" encoding="UTF-8"?>
<note>
<shorttitle>shorttitle_1</shorttitle>
<shorttitle>shorttitle_2</shorttitle>
<shorttitle>shorttitle_3</shorttitle>
<
浏览 2提问于2018-09-12得票数 2
1回答
熊猫星星之火
、、
我正在努力学习pyspark，而且在大多数情况下，我一直在使用使用数据集的pyspark.sql方法。然而，我对熊猫的火花api很感兴趣，因为我曾经广泛地使用过熊猫，但是我一直面临着运行时间的问题，特别是我习惯于在熊猫身上做的一些基本的事情。
例如，假设我使用的是ps_df ( pyspark.sql dataframe )和ps_pandas_df ( pyspark.pandas dataframe)：
import os
os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
from pyspark.sql import
浏览 7提问于2022-10-14得票数 0
1回答
我在这个数据中有一个星星之火的dataframe(input_dataframe)，数据，如下所示：
id value
1 a
2 x|y|z
3 t|u
我希望output_dataframe，让管道分隔的字段爆炸，它应该如下所示：
id value
1 a
2 x
2 y
2 z
3 t
浏览 1提问于2017-09-12得票数 0
我在dataframe中有一个列，它的字符串值如下
"Hardware part not present"
"Software part not present"
null
null
我希望拆分wrt“”，并且只将前2个字符串使用到新列，如果它是null，那么即使是新列值也应该为null。如何做到这一点？
所需结果
column New column
Hardware part not present Hardware part
Software part not present
浏览 5提问于2022-09-30得票数 0
1回答
我正在使用pyspark dataframe从每个行的数组中查找不同的计数:输入: col1 1,1,1 1,2,1,2
output:
1
3
2
I used below code but it is giving me the length of an array:
output:
3
3
4
please help me how do i achieve this using python pyspark dataframe.
slen = udf(lambda s: len(s), IntegerType())
count = Df.withColumn("Coun
浏览 10提问于2020-02-28得票数 1
回答已采纳
我想使用Python将Dataframe (pyspark.pandas.Dataframe)作为一个Excel文件保存在Azure数据湖Gen2上。我已经切换到pyspark.pandas.Dataframe，因为它是从Spark3.2开始推荐的。
有一种名为to_excel (，文档)的方法允许将文件保存到ADL中的容器中，但我在文件系统访问协议方面遇到了问题。在同一个类中，我使用了使用abfss的to_csv和to_parquet方法，我想在excel中使用相同的方法。
因此，当我尝试这样保存它时，我使用：
import pyspark.pandas as ps
# Omit the d
浏览 32提问于2022-05-11得票数 0
回答已采纳
1回答
我正在使用PySpark运行一些不同数据集的模拟，我想保存所有控制台输出(信息、警告等)。通过在代码中声明将包含日志输出的文本文件，以一种动态的方式对文本文件进行声明。代码将简单地在输入数据集上运行一些操作，我计划使用spark-sumbit运行代码。
这将允许我为单独的模拟保存单独的日志，其背后的想法是将日志文件名与输入数据集名称匹配。
如果不更改confs和其他Spark文件，这是可能的吗？
浏览 2提问于2016-06-09得票数 2
回答已采纳
我想处理从Kafka流到PySpark的日志数据并保存到拼图文件中，但我不知道如何将数据输入到Spark。请帮帮我谢谢。
浏览 68提问于2020-07-04得票数 0
我试图在Yarn框架上以客户端模式读取本地文件。我也无法在客户端模式下访问本地文件。
import os
import pyspark.sql.functions as F
from os import listdir, path
from pyspark import SparkConf, SparkContext
import argparse
from pyspark import SparkFiles
from pyspark.sql import SparkSession
def main():
spark = SparkSession \
.builder \
浏览 1提问于2018-10-18得票数 1
回答已采纳
希望有人能帮助我在Pyspark中做一个简单的情感分析。我有一个Pyspark dataframe，其中每一行都包含一个word。我还有一本常见stopwords的字典。 我想删除stopwords字典中word (行的值)所在的行。 输入： +-------+
| word |
+-------+
| the|
| food|
| is|
|amazing|
| and|
| great|
+-------+
stopwords = {'the', 'is', 'and'} 预期输出： +-------+
|
浏览 40提问于2021-11-12得票数 0
回答已采纳
如何解析嵌套列表中的JSON字符串以激发pyspark中的数据帧？
输入数据框架：
+-------------+-----------------------------------------------+
|url |json |
+-------------+-----------------------------------------------+
|https://url.a|[[1572393600000, 1.000],[1572480000000, 1.007]]|
|h
浏览 1提问于2021-02-08得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
