使用pyspark或spark SQL选择不在组中的所有或特定列
基础概念
Apache Spark 是一个快速、通用的大规模数据处理引擎，支持多种编程语言，包括 Python（通过 PySpark）和 SQL（通过 Spark SQL）。PySpark 和 Spark SQL 允许用户进行分布式数据处理和分析。
相关优势
- 分布式处理：Spark 可以在集群上分布式处理大量数据。
- 内存计算：Spark 支持将数据缓存在内存中，以提高处理速度。
- 多语言支持：除了 Scala，还支持 Java、Python 和 R。
- 丰富的 API：提供了丰富的数据处理和分析 API。
类型
- PySpark：Spark 的 Python API，允许用户使用 Python 进行数据处理。
- Spark SQL：Spark 的 SQL 接口，允许用户使用 SQL 语句进行数据处理。
应用场景
- 大数据分析：处理和分析大规模数据集。
- 机器学习：构建和训练机器学习模型。
- 实时数据处理：处理实时数据流。
问题描述
假设我们有一个 DataFrame，我们想要选择不在某个组中的所有列或特定列。
示例代码
假设我们有一个 DataFrame df，其中包含以下列：id, name, group。我们想要选择不在 group 为 A 的所有列。
使用 PySpark
from pyspark.sql import SparkSession
# 创建 SparkSession
spark = SparkSession.builder.appName("example").getOrCreate()
# 示例数据
data = [
(1, "Alice", "A"),
(2, "Bob", "B"),
(3, "Charlie", "A"),
(4, "David", "C")
]
# 创建 DataFrame
columns = ["id", "name", "group"]
df = spark.createDataFrame(data, columns)
# 选择不在 group 为 A 的所有列
result = df.filter(df.group != "A")
# 显示结果
result.show()使用 Spark SQL
from pyspark.sql import SparkSession
# 创建 SparkSession
spark = SparkSession.builder.appName("example").getOrCreate()
# 示例数据
data = [
(1, "Alice", "A"),
(2, "Bob", "B"),
(3, "Charlie", "A"),
(4, "David", "C")
]
# 创建 DataFrame
columns = ["id", "name", "group"]
df = spark.createDataFrame(data, columns)
# 注册 DataFrame 为临时视图
df.createOrReplaceTempView("people")
# 使用 SQL 语句选择不在 group 为 A 的所有列
result = spark.sql("SELECT * FROM people WHERE group != 'A'")
# 显示结果
result.show()参考链接
解决问题的思路
- 创建 SparkSession：这是使用 PySpark 或 Spark SQL 的第一步。
- 创建 DataFrame：使用示例数据创建 DataFrame。
- 过滤数据：使用
filter方法或 SQL 语句过滤掉不需要的数据。 - 显示结果：使用
show方法显示结果。
通过上述步骤，我们可以选择不在某个组中的所有列或特定列。
相关·内容
我已经创建了一个空的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
回答已采纳
2回答
我正在尝试访问蜂箱拼板表，并将其加载到Pandas数据帧中。我使用的是pyspark，我的代码如下所示：
import pyspark
import pandas
from pyspark import SparkConf
from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql import HiveContext
conf = (SparkConf().set("spark.driver.maxResultSize", "10g").setApp
浏览 0提问于2017-05-31得票数 1
当我阅读Spark/Hive文档时，将插入到带有列列表的表中，在Spark2.4和更早版本中不支持。
我有一个源表和一个目标表，它们有不同的列数和不同的列名，我需要复制它们。
这是否意味着我必须在PySpark中编写此代码以完成此工作，因为Spark将无法完成此任务？
示例：
input_table( cola, colb, colc, cold, cole)
output_table(fieldx, fieldy, fieldz)
在SQL中(假设关系数据库管理系统，如MS、PostgreSQL等)，我将执行以下操作：
insert into output_table(fieldx, fi
浏览 1提问于2019-01-09得票数 4
回答已采纳
我在测试文件夹中有多个测试文件。这个结构类似于这样的东西：
/test
----test_abc.py
----test_bcd.py
----test_cde.py
----conftest.py
conftest.py包含运行单元测试所必需的所有星火上下文初始化。我的问题是，我希望有一个test.py文件，它在内部触发所有的test_abc.py、test_bcd.py和test_cde.py。当我们处理python的utit_test模块时，它变得非常容易，但我不知道如何通过pytest模块来获得它。如果需要对这个问题作进一步的澄清，请告诉我。
conftest.py看起来如下所示：
i
浏览 2提问于2018-12-10得票数 4
我在PySpark 3.0.1的中找到了用于jdbc函数的文档，它说：
列-将用于分区的数值、日期或时间戳类型列的名称。
我认为它接受一个datetime列来划分查询。
因此，我在EMR-6.2.0 (PySpark 3.0.1)上尝试了这个方法：
sql_conn_params = get_spark_conn_params() # my function
sql_conn_params['column'] ='EVENT_CAPTURED'
sql_conn_params['numPartitions'] = 8
# sql_conn_p
浏览 6提问于2021-02-06得票数 2
回答已采纳
1回答
我希望能够简洁地指定我添加到列表理解中的列。
列表理解的基本结构非常简单，其中spark_sql_df是星星之火数据，列是数据模型中列的实际名称。
def return_list():
# creation and munging of spark_sql_df
my_list = [int(row.column) for row in spark_sql_df.collect()]
return my_list
然而，由于我需要搜索多个数据，然后比较结果，以揭示丢失的数据，我想要一种能够循环遍历列的方法，这样我就不必重复这些代码多次了。
简单地说
浏览 5提问于2020-06-30得票数 0
回答已采纳
我在Amazon AWS上运行了Spark 1.6。我试图运行这段代码，其中我通过传入主节点url来创建SparkContext，但是我得到了连接被拒绝的错误。我当前的代码是：
import fnmatch
import os
import sys
from os import system
from sys import argv
from pyspark import SparkContext
from pyspark.sql import SQLContext
def test(master_url):
sc = SparkContext(master=master_url,
浏览 2提问于2016-03-29得票数 0
目前，我可以通过RDD使用以下代码将文本文件解析为DataFrame：
def row_parse_function(raw_string_input):
# Do parse logic...
return pyspark.sql.Row(...)
raw_rdd = spark_context.textFile(full_source_path)
# Convert RDD of strings to RDD of pyspark.sql.Row
row_rdd = raw_rdd.map(row_parse_function).filter(bool)
# Conver
浏览 6提问于2016-09-02得票数 0
回答已采纳
2回答
我用docker创建了一个Hive容器，我创建了poke表，当我执行Select (一个after ()函数)时出现了这个错误。
守则：
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession, HiveContext
spark = SparkSession \
.builder \
.appName("Python Spark SQL Hive integration example") \
.config("hive.metasto
浏览 1提问于2018-04-25得票数 0
1回答
我正在尝试使用MongoDB连接器来使用PySpark。但是，仅创建PySpark会话就需要大约20到25秒，这会影响服务的性能。我还给出了用来创建spark会话的代码片段。有没有人能建议我怎么让它更快？ from pyspark.sql import SparkSession
my_spark = SparkSession \
.builder \
.appName("myApp") \
.config("spark.mongodb.input.uri", "mongodb://localhost:27
浏览 18提问于2020-04-14得票数 2
2回答
我正在运行以下代码段，
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import ntile
spark = SparkSession.builder.appName('SparkByExamples.com').getOrCreate()
simpleData = (("James", "Sales", 3000), \
("Michael
浏览 4提问于2021-04-02得票数 1
回答已采纳
我正在尝试加载拼花的文件，在S3到红移使用胶水作业。当我第一次运行Glue Job时，它正在创建表和加载数据，但当第二次通过更改1列的数据类型运行时，作业并没有失败，而是在Redshift中创建了新列并附加了数据。
例如:在这里，我更改了整型数字的数据类型
FileName **abc**
Code,Name,Amount
'A','XYZ',200.00
FileName **xyz**
Code,Name,Amount
'A','XYZ',200.00
在Redshift中
Output after processing b
浏览 3提问于2020-03-02得票数 2
我试图在Spark1.6中列出使用HiveContext的所有数据库，但它只给了我默认的数据库。
from pyspark import SparkContext
from pyspark.sql import SQLContext
sc = SparkContext.getOrCreate()
from pyspark.sql import HiveContext
sqlContext = HiveContext(sc)
sqlContext.sql("SHOW DATABASES").show()
+-------------+
| result|
+-----
浏览 4提问于2019-06-24得票数 0
回答已采纳
我已经使用以下代码创建了数据框： import pyspark
from pyspark.sql import functions as F
sc = pyspark.SparkContext()
spark = pyspark.sql.SparkSession(sc)
data = [('A', 'B', 1), ('A', 'B', 2), ('A', 'C', 1)]
columns = ['Column1', 'Col
浏览 12提问于2020-06-15得票数 0
1回答
我试图使用Spark2.2在DataFrame中打印每个分区中的总元素
from pyspark.sql.functions import *
from pyspark.sql import SparkSession
def count_elements(splitIndex, iterator):
n = sum(1 for _ in iterator)
yield (splitIndex, n)
spark = SparkSession.builder.appName("tmp").getOrCreate()
num_parts = 3
df = spa
浏览 0提问于2018-03-25得票数 6
2回答
我正试图在我的spark application.But中从Postgresql中获取数据，现在我很困惑如何在我的码头映像中安装postgresql驱动程序。我还尝试将postgresql安装为apt-get安装命令，如下所述(Dockerfile)。
Dockerfile：
FROM python:3
ENV SPARK_VERSION 2.3.2
ENV SPARK_HADOOP_PROFILE 2.7
ENV SPARK_SRC_URL https://www.apache.org/dist/spark/spark-$SPARK_VERSION/spark-${SPARK_VERSI
浏览 1提问于2018-12-25得票数 1
回答已采纳
如果我有一个包含arrays的Spark，我可以通过DataFrame在这些数组上使用Python List方法吗？如何将Spark DataFrame array<double>转换为Python列表？
下面是一个示例，其中包含一些UDF。我不确定为什么使用最大值可以，但是使用len就不行了。最后，我想用原始数组列中的采样值创建一个新列。这也会得到一个关于期望两个参数的错误，如果你能帮助的话也会得到加分！
我有以下Spark DataFrame
from pyspark.sql.functions import udf
from pyspark.sql import Row
fr
浏览 1提问于2016-10-28得票数 4
回答已采纳
我有一个到外部数据库的SparkSQL连接：
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.getOrCreate()
如果我知道一个表的名称，它就很容易查询。
users_df = spark \
.read.format("jdbc") \
.options(dbtable="users", **db_config) \
.load
浏览 2提问于2017-03-19得票数 11
回答已采纳
2回答
我是新来的火花和蜂巢。我需要了解当蜂箱表在星火中被查询时后面发生了什么。我正在使用PySpark
例如：
warehouse_location = '\user\hive\warehouse'
from pyspark.sql import SparkSession
spark =SparkSession.builder.appName("Pyspark").config("spark.sql.warehouse.dir", warehouse_location).enableHiveSupport().getOrCreate()
DF = s
浏览 0提问于2018-05-07得票数 0
回答已采纳
1回答
如何计算或管理火花放电中的流数据
、、、、
我想从流数据中计算数据，然后发送到网页。例如，：我将在流数据中计算TotalSales列的和。，但它在summary = dataStream.select('TotalSales').groupby().sum().toPandas()上出错，这是我的代码。
import os
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType
from pyspark.sql.functions import *
spark = SparkSession.builder.appName
浏览 0提问于2020-06-02得票数 0
回答已采纳
1回答
我正在尝试使用pyspark将数据从每天的批处理发送到Kafka主题，但目前我收到了以下错误：
文件跟踪(最近一次调用)：文件""，第5行，文件"/usr/local/rms/lib/hdp26_c5000/spark2/python/pyspark/sql/readwriter.py"，第548行，保存self._jwrite.save()文件"/usr/local/rms/lib/hdp26_c5000/spark2/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py"，第1133行，在 c
浏览 4提问于2020-11-18得票数 1
1回答
如何在AWS胶中使用火花包？
、、、、
我想在AWS中使用Datastax的。如果我在本地运行pyspark，我的命令看起来就像
path/to/spark-3.0.1-bin-hadoop2.7/bin/spark-submit \
--conf spark.cassandra.connection.host=XXX \
--conf spark.cassandra.auth.username=XXX \
--conf spark.cassandra.auth.password=XXX \
--packages com.datastax.spark:spark-cassandra-connector_2.12:2.5.1 \
~/
浏览 8提问于2020-09-16得票数 2
1回答
这是Google 到我的csv文件(<100 kb，别担心)
当我用Excel打开它时，一切都很好
但是当我和PySpark一起读它的时候，它只是搞砸了。我认为这是因为Body列，它是一个长而复杂的HTML字符串。你们知道怎么修吗？我已经更改了分隔符选项，但它没有工作。
import pyspark
from pyspark.sql import *
from pyspark.sql.functions import *
spark = SparkSession.builder.master("local[*]").getOrCreate()
questio
浏览 8提问于2022-11-15得票数 0
回答已采纳
1回答
我已经为python3.6安装了pycountry包。我使用以下命令启动PySpark外壳：
pyspark --driver-memory 50g --executor-memory 10g --conf "spark.pyspark.python=/usr/bin/python3.6" --conf "spark.pyspark.driver.python=/usr/bin/python3.6"
我定义了这个UDF：
@F.udf
def get_phone_number_country_iso(phone_number):
try:
浏览 0提问于2021-01-31得票数 1
我正在emr-5.4.0集群上运行齐柏林飞艇0.7.0。我将使用默认设置启动群集。%spark.dep解释器不通过EMR进行配置。
我从下面编辑了文件/etc/zeppelin/conf/interpreter.json：
"2ANGGHHMQ": {
"id": "2ANGGHHMQ",
"name": "spark",
"group": "spark",
"properties": {
"spark.yarn.jar": &
浏览 3提问于2017-03-27得票数 4
我试图将爆炸的列添加到dataframe中：
from pyspark.sql.functions import *
from pyspark.sql.types import *
# Convenience function for turning JSON strings into DataFrames.
def jsonToDataFrame(json, schema=None):
# SparkSessions are available with Spark 2.0+
reader = spark.read
if schema:
reader.schema(s
浏览 9提问于2022-12-02得票数 1
这是我的密码。
%spark.pyspark
jdbc_write(spark, spark.sql("""
SELECT
Global_Order_Number__c
, Infozeile__c
FROM STAG.SF_CASE_TRANS
"""), JDBC_URLS['xyz_tera_utf8'], "DEV_STAG.SF_CASE", "abc", "1234")
我想排除Infozeile__c字段中不是A、A
浏览 5提问于2022-02-14得票数 1
回答已采纳
我用Pycharm IDE编写了以下非常简单的python脚本
from pyspark.sql import SparkSession
from pyspark.sql import Row
from pyspark.sql import SQLContext
from pyspark.sql.types import LongType, FloatType,IntegerType,StringType,DoubleType
from pyspark.sql.functions import udf
from pyspark.sql import functions as F
from py
浏览 20提问于2019-11-12得票数 0
我有一个非常简单的脚本来持久化带有MongoDB中两列的数据文件：
from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
from pyspark.sql.types import *
from pyspark.sql.functions import col, udf
from datetime import datetime
sparkConf = SparkConf().setMaster("local").setAppName("Wiki-Analyzer
浏览 4提问于2017-03-30得票数 0
回答已采纳
我以以下方式创建了一个DataFrame：
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.getOrCreate()
df = spark.read.csv("train.csv", header=True)
我的DataFrame的架构如下：
root
|-- PassengerId: string (nullable = true)
|-- S
浏览 3提问于2018-10-18得票数 0
回答已采纳
我正尝试在运行Linux Mint的本地计算机上使用Jupyter Notebook初始化PySpark集群。我正在关注。当我尝试创建一个SparkSession时，我得到一个spark-submit不存在的错误。奇怪的是，这和我在不包含sudo的情况下获取spark-shell版本时得到的错误是一样的。
spark1 = SparkSession.builder.appName('Test').getOrCreate()
FileNotFoundError: [Errno 2] No such file or directory: '~/Spark/spark-3.
浏览 4提问于2021-07-12得票数 1
1回答
列中的搜索值
、、
我想搜索列是否包含值。 import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.types import *
import pandas as pd
df_init = pd.DataFrame({'id':['1', '2'], 'val':[100, 200]})
spark = SparkSession.builder.appName('pandasToSparkDF').getOrCreate()
mySchema
浏览 16提问于2020-09-29得票数 1
回答已采纳
使用PySpark SQL并给定3列，我想创建一个额外的列，该列将其中的两列分开，第三列是ID列。 df = sqlCtx.createDataFrame(
[
(1, 4, 2),
(2, 5, 2),
(3, 10, 4),
(4, 50, 10)
],
('ID', 'X', 'Y')
) 这是所需的输出： +----+----+----+---------------------+
| ID | x | y | z (expected resul
浏览 27提问于2019-05-10得票数 3
回答已采纳
我可以在Azure Databricks中使用PySpark执行一个简单的SQL语句，但是我想执行一个存储过程。下面是我尝试过的PySpark代码。
#initialize pyspark
import findspark
findspark.init('C:\Spark\spark-2.4.5-bin-hadoop2.7')
#import required modules
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession
from pyspark.sql impo
浏览 5提问于2020-02-23得票数 5
2回答
我有一个pyspark函数，看起来像这样。\
spark.sql("select count(*) from student_table where student_id is NULL") \
spark.sql("select count(*) from student_table where student_scores is NULL") \
spark.sql("select count(*) from student_table where student_health is NULL")
我得到的结果看起来像\
+-------+|
浏览 0提问于2021-11-07得票数 0
1回答
除了重新分区(这减慢了处理速度)，还有其他方法将所有1mb文件合并成多个大文件吗？
在500 on的数据上运行spark代码，在100个执行器上运行24个核，但将它们保存到包含128 on的大型文件中。现在，它正在保存每个文件1MB。
spark.sql("set pyspark.hadoop.hive.exec.dynamic.partition=true")
spark.sql("set pyspark.hadoop.hive.exec.dynamic.partition.mode=nonstrict")
spark.sql("set hive.
浏览 0提问于2019-06-27得票数 2
回答已采纳
我有两个DataFrames，我想对这两个DataFrame的每一列应用distance.euclidean(df1.select(col),df2.select(col))。 示例： from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *
spark = SparkSession.builder.getOrCreate()
df1 = spark.createDataFrame([(1,10),(2,13)],["A
浏览 9提问于2020-07-11得票数 0
我有一个由调度程序触发的AWS胶水中的ETL作业。我的ETL语言是python。我正在尝试-在s3桶中编写查询结果。为此，我使用了sparkSql。此作业在由调度程序触发时失败，但在手动运行时成功。它正在抛出一个列(eventdate)的错误，该列在spark中可用。
下面是日志。
Traceback (most recent call last):
File "script_2018-06-22-11-10-05.py", line 48, in <module>
error_report_result_df = spark.sql(sql_query)
File
浏览 0提问于2018-06-22得票数 3
回答已采纳
我正在尝试迭代一个大型Pyspark Dataframe列中的所有不同值。当我尝试使用.distinct().collect()执行此操作时，即使只有两个不同的值，它也会发出“任务太大”警告。警告消息：
20/01/13 20:39:01 WARN TaskSetManager: Stage 0 contains a task of very large size (201 KB). The maximum recommended task size is 100 KB.
下面是一些示例代码：
from pyspark.sql import SparkSession
spark = Spar
浏览 1提问于2020-01-14得票数 1
1回答
我已经安装和设置了和集成。通过使用spark-shell / pyspark，我还遵循并实现了创建Hive表，加载数据，然后正确选择。
然后转到下一步，设置。通过使用hive / beeline，我还可以创建Hive表，加载数据，然后进行适当的选择。蜂箱在纱线/火花上正确地执行。我怎么知道它起作用了？hive外壳显示以下内容：
hive> select sum(col1) from test_table;
....
Query Hive on Spark job[0] stages: [0, 1]
Spark job[0] status = RUNNING
---------------
浏览 0提问于2020-02-23得票数 0
回答已采纳
我想转换一个大的spark数据帧到Pandas超过1000000行。我尝试使用以下代码将spark数据帧转换为Pandas数据帧：
spark.conf.set("spark.sql.execution.arrow.enabled", "true")
result.toPandas()
但是，我得到了错误：
TypeError Traceback (most recent call last)
/usr/local/lib/python3.6/dist-packages/pyspark/sql/da
浏览 3提问于2018-07-04得票数 7
我使用pyspark流从kafka获取数据，结果是一个dataframe，当我将dataframe转换为rdd时，它出错了： Traceback (most recent call last):
File "/home/docs/dp_model/dp_algo_platform/dp_algo_core/test/test.py", line 36, in <module>
df = df.rdd.map(lambda x: x.value.split(" ")).toDF()
File "/home/softs/spark-2.4.3-
浏览 56提问于2020-01-06得票数 3
我正在尝试加入两个数据处理程序：
df1，列：
document_nbr, line_id, product_id, product_size, reference_nbr, local_cd, invoice_local_cost, invoice_delivery_id, created_by_id, transaction_ind, etl_tmst, record_created_tmst, record_updated_tmst, source_id, line_number, etl_date
加入df1和df2：
df2栏：
document_nbr, line_id, vari
浏览 1提问于2022-05-10得票数 -1
2回答
SparkSQL查询数据帧
、、、、
我将一个熊猫数据帧转换成spark sql表。我是SQL的新手，想从表中选择关键的'code‘。 查询 sqlContext.sql("""SELECT `classification` FROM psyc""").show() 查询响应 +--------------------+
| classification|
+--------------------+
|[{'code': '3297',...|
|[{'code': '3410',...|
|[{&
浏览 13提问于2020-01-29得票数 0
回答已采纳
1回答
我有一个包含纯文本的字符串类型列的dataframe，我想使用pyspark.sql.functions.udf (或pyspark.sql.functions.UserDefinedFunction?)修改这个列。
我在OSX10.11.4上使用Python2.7、Pyspark 1.6.1和Flask 0.10.1。
当我使用lambda表达式时，它似乎工作得很好：
@spark.route('/')
def run():
df = ... # my dataframe
myUDF = udf(lambda r: len(r), IntegerType()
浏览 10提问于2016-05-24得票数 0
回答已采纳
1回答
如何从字符串值中获取数字代码？
、、、、
我已经使用pyspark.sql将csv文件导入spark，并通过以下方式将其注册为临时表：
import pyspark
from pyspark.sql import SQLContext
sc = pyspark.SparkContext()
from pyspark.sql import HiveContext
sqlCtx= HiveContext(sc)
spark_df = sqlCtx.read.format('com.databricks.spark.csv').options(header='true', inferschema='
浏览 3提问于2017-01-26得票数 0
我正在使用ApacheSpark1.6.2
我有一个.csv数据，它包含大约800万行，我想将它转换为DataFrame
但是，我必须首先将它转换为RDD来进行映射，以获得我想要的数据(列)。
映射RDD很好，但是当涉及到将RDD转换为DataFrame时，火花抛出一个错误。
Traceback (most recent call last):
File "C:/Users/Dzaky/Project/TJ-source/source/201512/final1.py", line 38, in <module>
result_iso = input_i
浏览 3提问于2017-01-14得票数 4
回答已采纳
我之前发布了这个问题，并得到了一些使用PySpark的建议。
下面的压缩文件()包含一个名为data的文件夹，其中包含大约130,000个csv文件。我想把它们合并到一个单独的数据帧中。我有16 of的RAM，当我访问前几百个文件时，我一直在耗尽RAM。这些文件的总大小只有300-400MB的数据。
如果您打开任何csv文件，您可以看到它们都具有相同的格式，第一列用于日期，第二列用于数据系列。
所以现在我改用PySpark，但是我不知道连接所有文件的最有效的方法是什么，对于pandas数据帧，我会像这样连接单个帧的列表，因为我想让它们在日期上合并：
bigframe = pd.concat(
浏览 14提问于2020-02-17得票数 2
回答已采纳
我目前正在使用databricks，并且有一个包含20+列的增量表。基本上，我需要从每一行的1列中提取一个值，将其发送到一个返回两个值/列的api，然后创建另一个26列，以将这些值合并回原始增量表。所以输入是28列，输出是28列。目前，我的代码如下所示： from pyspark.sql.types import *
from pyspark.sql import functions as F
import requests, uuid, json
from pyspark.sql import SparkSession
from pyspark.sql import DataFrame
f
浏览 74提问于2021-10-18得票数 2
回答已采纳
1回答
左反团员
、、、
我有一个dataframe，它有两个列a和b，其中b列中的值是a列中值的a子集。例如：
df
+---+---+
| a| b|
+---+---+
| 1| 2|
| 1| 3|
| 2| 1|
| 3| 2|
+---+---+
我想生成一个包含a和anti_b列的数据格式，其中anti_b列中的值是来自a列的任何值，这样a!=anti_b和行(a,anti_b)就不会出现在原始的数据格式中。因此，在上面的数据中，结果应该是：
anti df
+---+------+
| a|anti_b|
+---+------+
| 3| 1|
| 2| 3|
浏览 2提问于2019-11-18得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
