我无法使用Dataframe中的选择所需的列。如果我从df_ord DataFrame中选择一列，则结果将显示来自df_ord的一列，并对df_od_item数据的列进行重命名，这些列是不正确的。请参阅所附的截图。
另外，当我从两个Dataframe中选择多列时，我得到了一个错误。请协助。
Py4JJavaError Traceback (most recent call last)
/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py in deco(*a,
浏览 4提问于2020-05-23得票数 1
回答已采纳
我有一个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
我正在使用Pyspark，并且正在使用spark.sql编写一个查询。我希望从在其他地方声明的数组中选择值，以避免再次键入所有行的名称。
这是我的尝试，但这不起作用。
array_fields = ["cat", "dog"]
ans= spark.sql("""select {} from <table_name>.format(",".join[array_fields]) """)
我也试过
ans= spark.sql("""select {} fr
浏览 1提问于2020-03-11得票数 0
我刚接触Hive和spark sql技术，我曾经尝试过spark sql中的limit子句。但它只支持从零开始到特定limit.But的特定限制，我想检索从特定起始点到特定结束点的行，您可以建议一种方法来实现这一点。
Query1 :: SELECT * FROM `Emp` LIMIT 10; - this query supports in both sql and spark sql
但
Query2 :: SELECT * FROM `Emp` LIMIT 10,20; - to retrive rows from 10 to 20 supports in sql, but not
浏览 1提问于2016-05-12得票数 0
当我提交复杂的join SQL查询时，通常会给一个或2个操作数更短的名称来澄清我的意图，例如，以下两个查询：
SELECT *
FROM transactions
JOIN accounts ON transactions.cardnumber=accounts.cardnumber
和
SELECT *
FROM transactions AS left
JOIN accounts ON left.cardnumber=accounts.cardnumber
应该有同样的效果。
我已经在Spark1.6.3中测试了这两种查询，而且这两种查询都能工作。但是，在我移到Spark2.2.1之后，第
浏览 3提问于2018-01-08得票数 1
回答已采纳
我试图从包含列表中单词的数据行中提取:下面是粘贴我的代码：
from pyspark.ml.feature import Tokenizer, RegexTokenizer
from pyspark.sql.functions import col, udf
from pyspark.sql.types import IntegerType
from pyspark.sql.types import IntegerType
import findspark
findspark.init()
conf = SparkConf().setAppName("nlp").setMast
浏览 3提问于2021-06-16得票数 2
回答已采纳
我尝试在Spark SQL中重新创建SQL查询。通常我会像这样插入到一个表中：
INSERT INTO Table_B
(
primary_key,
value_1,
value_2
)
SELECT DISTINCT
primary_key,
value_1,
value_2
FROM
Table_A
WHERE NOT EXISTS
(
SELECT 1 FROM
Table_B
WHERE
Table_B.primary_key = Table_A.primary_key
);
Spark SQL很简单，我可以将数据从T
浏览 1提问于2017-04-19得票数 0
我有以下数据帧： dataframe1
+-----------------------+
|ID |
+-----------------------+
|[10,80,60,] |
|[20,40,] |
+-----------------------+ 和另一个数据帧： dataframe2
+------------------+----------------+
|ID_2 | name |
+------------------+-------
浏览 76提问于2020-08-15得票数 0
2回答
我使用的是pyspark (Apache Spark)的DataFrame接口，遇到了以下问题：
当我连接来自相同源DataFrame的两个DataFrames时，得到的DF将分解为大量的行。一个简单的例子：
我从磁盘加载一个包含n行的DataFrame：
df = sql_context.parquetFile('data.parquet')
然后，我从该源文件创建了两个DataFrames。
df_one = df.select('col1', 'col2')
df_two = df.select('col1', 'c
浏览 1提问于2015-04-21得票数 6
1回答
一般进口问题
、、、
我成功地在我的机器中安装了火花和火花，添加了路径变量等，但仍然面临导入问题。
这是代码：
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.config("spark.hadoop.hive.exec.dynamic.partition", "true") \
.config("spark.hadoop.hive.exec.dynamic.partition.mode",
浏览 5提问于2021-10-29得票数 0
回答已采纳
当行中的某些列值与Source表中的相同列值匹配时，我希望更新databricks中的目标Delta表。
问题是当源表中有多个行与目标Delta表中的一行匹配时。
在这种情况下，源表中两个或两个以上行的主键与delta表中的一行主键匹配。我试图复制以下场景：
sql="""
MERGE INTO """ + targetTable + """ TGT USING """ + tempView + """ SRC ON TGT.id = SRC.id and T
浏览 2提问于2020-06-08得票数 0
回答已采纳
我对Spark2.1中的join有一个问题。Spark (错误的？)选择广播散列join，尽管该表非常大(1400万行)。然后作业崩溃，因为没有足够的内存，Spark以某种方式尝试将广播片段持久存储到磁盘，这将导致超时。
所以，我知道有一个查询提示可以强制广播连接(org.apache.spark.sql.functions.broadcast)，但是有没有办法强制另一个连接算法呢？
我通过设置spark.sql.autoBroadcastJoinThreshold=0解决了我的问题，但我更喜欢另一种更细粒度的解决方案，即不全局禁用广播加入。
浏览 32提问于2018-01-08得票数 7
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
我有两个数据帧df1和df2。它们都有以下架构：
|-- ts: long (nullable = true)
|-- id: integer (nullable = true)
|-- managers: array (nullable = true)
| |-- element: string (containsNull = true)
|-- projects: array (nullable = true)
| |-- element: string (containsNull = true)
df1是从avro文件创建的，而df2是从等效的拼图文件创建的。但是
浏览 0提问于2016-07-29得票数 16
回答已采纳
1回答
我正在做一个从传入的结构化数据(在HDFS上的文件夹中以CSV文件的形式)中删除重复记录的用例。为了尝试此用例，我使用文件选项编写了一些示例代码，以查看是否可以从复制到文件夹(HDFS)的CSV中存在的记录中删除重复项。
在代码片段下面找到：
import org.apache.spark.sql.functions._
import org.apache.spark.sql.SparkSession
import spark.implicits._
import org.apache.spark.sql.types.{StructType, StructField, StringType,
浏览 0提问于2019-03-22得票数 0
1回答
当我从s3桶将数据加载到pyspark中时，进行一些操作(连接、联合)，然后尝试覆盖前面读取的相同路径(' data /csv/')。我得到了一个错误：
py4j.protocol.Py4JJavaError: An error occurred while calling o4635.save.
: org.apache.spark.SparkException: Job aborted.
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.sca
浏览 0提问于2019-06-12得票数 0
与一样，SparkSQL抱怨标量子查询。
因此，当不能通过查看SQL语句(不查看数据)就能100%确定子查询只返回一行时，就会抛出这个异常。如果您确信您的子查询只给出一行，那么您可以使用下列聚合标准函数之一，因此Spark很高兴：
firstavgmaxmin
如何使下面的SQL查询工作？
SELECT
prerequisite AS prerequisite,
(SELECT e.description FROM course e WHERE e.course_no = c.prerequisite) as course_name,
COUNT(*) as cnt
浏览 12提问于2022-02-19得票数 0
我一直在尝试将1个DataFrame附加到Scala中的另一个DF上。本例中的追加操作只是将一个相同大小的新列添加到现有列中-不涉及键匹配。两个DataFrames的形状相同(仅5行1列)。
scala> val coefficients = lrModel.coefficients.toArray.toSeq.toDF("coefficients")
coefficients: org.apache.spark.sql.DataFrame = [coefficients: double]
scala> coefficients.show()
+---------
浏览 8提问于2017-08-05得票数 0
回答已采纳
我有一个表，在一个数据库中。有一列" id“，并且每个id都有多行。我想随机挑选10个I，然后用这些I加载所有行。下面的代码是有效的，并给出了一个包含10个随机ids的表。这很好。 distinct_ids = spark.sql(f""" (SELECT DISTINCT id FROM {database_name}.{orig_table_cl_name} ORDER BY RAND() LIMIT 10) 但是现在我不知道如何对原始表进行内连接，以获得包含正确I的所有数据……我试过了： distinct_ids = spark.sql(
f"
浏览 10提问于2021-11-10得票数 0
回答已采纳
2回答
首先，我构建了scala应用程序，使用这一行代码从apache中的mysql表中读取数据。
val spark = SparkSession.builder().master("local").appName("Fuzzy Match Analysis").config("spark.sql.warehouse.dir","file:///tmp/spark-warehouse").getOrCreate()
import spark.implicits._
var df = spark.read.format("jdbc
浏览 2提问于2017-10-02得票数 1
我运行了一个简单的Spark程序来获取数据从Hive到session使用spark。
scala> spark.sql("select count(1) from firsthivestreamtable").show(100,false)
+--------+
|count(1)|
+--------+
|36 |
+--------+
运行insert语句在Hive表中插入9条新记录(直接在Hive控制台上)。验证Hive表有正确插入的其他行。
hive> select count(1) aa from firsthivestreamtable;
To
浏览 4提问于2019-05-24得票数 1
回答已采纳
我需要了解如何使用Java在Spark SQL中基于single从数据框中删除重复行。
与普通SQL中一样，ROW_NUMBER () OVER (PARTITION BY col BY Col DESC)。如何将此步骤转换为Java中的Spark SQL？
浏览 11提问于2017-07-25得票数 1
我在Databricks集群中有一个包含500万行的星星之火数据。我想要的是缓存这个火花数据，然后应用.count()，以便下一个操作运行得非常快。我过去做过两万行，而且它能工作。然而，在我尝试这样做的过程中，我遇到了以下悖论：
Dataframe创建
步骤1:从Azure数据湖存储帐户读取800万行
read_avro_data=spark.read.format("avro").load(list_of_paths) #list_of_paths[0]='abfss://storage_container_name@storage_account_name.dfs.
浏览 0提问于2020-06-01得票数 3
回答已采纳
如何使用Spark SQL过滤器作为不区分大小写的过滤器。
例如：
dataFrame.filter(dataFrame.col("vendor").equalTo("fortinet"));
只返回'vendor'列等于'fortinet'的行，但我希望'vendor'列等于'fortinet'或'Fortinet'或'foRtinet'的行...
浏览 6提问于2016-01-20得票数 16
回答已采纳
1回答
我有一个场景，其中有两个工作表(Csv)。为它创建了两个表。当有一个良好的数据时，我可以从第二个表中用一个值( id统计值)映射它。如果我有坏的数据，我应该再次用id统计值映射它(但是另一个值)。但是，我无法在spark中使用not。我得到的错误如下：
不匹配输入‘从’期待{，‘哪里’，‘组’，‘订单’，‘拥有’，‘限制’，‘横向’，‘窗口’，‘联合’，‘除’，‘相交’，‘排序’，‘群集’，‘分发’}(第1行，pos 386)
at org.apache.spark.sql.catalyst.parser.ParseException.withCommand(ParseDriver.scala
浏览 3提问于2017-11-06得票数 1
如果我将以下代码放入一个扩展App特性的对象中，并使用Idea的run命令运行它，那么下面的代码就没有问题。
但是，当我尝试从工作表中运行它时，我会遇到以下情况之一：
1-如果第一行在场，我得到：
不可序列化的任务: java.io.NotSerializableException:A$A34$A$A34
2-如果第一行被注释掉，我得到：
无法为内部类A$35$A$35$A12生成编码器，而无法访问定义该类的作用域。
//First line!
org.apache.spark.sql.catalyst.encoders.OuterScopes.addOuterScope(th
浏览 12提问于2017-04-05得票数 4
我想要更新spark-SQL中的表中的一行，那么我该怎么做呢？例如，我的原始查询是，从学生集标记更新= 56，其中id =1；
如何在spark-SQL中做到这一点？spark-SQL中不支持As update。请在这方面提供帮助。
浏览 26提问于2016-01-14得票数 2
在执行看似简单的火花sql筛选工作时，我得到了一个例外：
someOtherDF
.filter(/*somecondition*/)
.select($"eventId")
.createOrReplaceTempView("myTempTable")
records
.filter(s"eventId NOT IN (SELECT eventId FROM myTempTable)")
知道我怎么解决这个问题吗？
注意：
someOtherDF包含过滤后的~1M到5M行，e
浏览 4提问于2017-04-27得票数 3
回答已采纳
在我们的一个Pyspark作业中，我们有一个场景，我们在一个大的数据帧和相对较小的数据帧之间进行连接，我相信spark正在使用广播连接，我们遇到了以下错误 org.apache.spark.SparkException: Cannot broadcast the table that is larger than 8GB: 8 GB
at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec$$anonfun$relationFuture$1$$anonfun$apply$1.apply(BroadcastExch
浏览 752提问于2020-04-22得票数 0
我正在尝试使用SparkSession从hive表创建数据帧，如下所示。创建后，我将通过Ids列表来筛选行。
val myDF = spark.sql("select * from myhivetable")
val someDF = mfiDF.where(mfiDF("id").isin(myList:_*))
有没有一种方法可以查询配置单元表，如下所示：
val myDF = spark.sql("select * from myhivetable").where (("id").isin(myList:_*))
当我尝试
浏览 0提问于2018-09-26得票数 1
我有两个数据文件如下：
course.txt
id,course
1,Hadoop
2,Spark
3,HBase
5,Impala
Fee.txt
id,amount
2,3900
3,4200
4,2900
我需要列出所有的课程信息和他们的费用：
sqlContext.sql("select c.id, c.course, f.amount from course c left outer join fee f on f.id = c.id").show
+---+------+------+
| id|course|amount|
+---+------+---
浏览 0提问于2018-03-10得票数 3
回答已采纳
2回答
我认为这个问题类似于其他一些问题，但它没有被问到。
在Spark中，我们如何在删除重复列的情况下运行SQL查询？
例如，在spark上运行的SQL查询
select a.* from a
left outer join
select b.* from b
on a.id = b.id
在这种情况下，如何删除重复的列b.id？
我知道我们可以在Spark中使用其他步骤，比如提供遗憾或重命名列，但是有没有一种更快的方法来简单地通过编写SQL查询来删除重复的列？
浏览 8提问于2017-09-10得票数 1
1回答
我使用Spark获得了100行的结果集。我希望得到从第6行开始到第15行的最终结果。在SQL中，我们使用OFFSET跳过行，比如OFFSET 5 LIMIT 10用于从第6行到第15行。在Spark中，如何实现相同的结果？
浏览 1提问于2017-03-04得票数 5
3回答
火花:与阵列连接
、、
我需要将一个带有string列的dataframe连接到一个字符串数组中，这样如果数组中的值匹配，行就会连接起来。
我试过了，但我想这不是支持。还有别的办法吗？
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("test")
val spark = SparkSession.builder().config(sparkConf).g
浏览 1提问于2017-08-07得票数 10
回答已采纳
我有两个小表要做完全的外连接，如下所示，我认为它应该使用广播连接，但它选择了排序合并连接，我想知道为什么。 test("SparkTest 0461") {
val spark = SparkSession.builder().master("local").appName("SparkTest0460").getOrCreate()
import spark.implicits._
val data1 = Seq((1, 2), (1, 7), (3, 6), (5, 4), (1, 10), (6, 7), (
浏览 40提问于2019-03-03得票数 0
回答已采纳
我有两个数据帧：
val df1 = List(("Sid", 1), ("Teni", 2), ("Bob", 3), ("Spark", 4), ("Hbase", 5)).toDF("name", "value")
val df2 = List(("Sidhartha", 1), ("Tammineni", 2), ("Bobby", 3), ("Spark", 4)).toDF("name", &
浏览 0提问于2021-04-14得票数 0
1回答
自从上次使用spark太久了，我再次使用Spark3.1，下面是我的问题:我还有2000万行加入400米行，原始代码是：
times= [50000,20000,10000,1000]
for time in times:
join = (df_a.join(df_b,
[
df_a["a"] == df_b["a"],
(unix_times
浏览 1提问于2021-08-28得票数 1
3回答
我是Spark的新手，需要使用PySpark或Spark Sql将以下输入数据帧转置为所需的输出df (从行到列)的帮助。
输入数据帧-
A B C D
1 2 3 4
10 11 12 13
......
........
所需的输出(转置)数据
A 1
B 2
C 3
D 4
A 11
B 12
C 13
D 14
....
......
如果我可以根据我们的要求旋转输入数据(列)，那就更好了。
浏览 1提问于2020-11-27得票数 3
当我在Databricks/PySpark中运行查询时，我得到以下错误：
org.apache.spark.SparkException: Could not execute broadcast in 300 secs. You can increase the timeout for broadcasts via spark.sql.broadcastTimeout or disable broadcast join by setting spark.sql.autoBroadcastJoinThreshold to -1
如何在Databricks notebook中以编程方式(Pytho
浏览 54提问于2020-11-13得票数 0
回答已采纳
我是Spark SQL的新手。我们正在将数据从SQL server迁移到Databricks。我使用的是SPARK SQL。你能建议如何在SPARK sql中为下面的日期函数实现以下功能吗？我可以看到，在spark sql中，datediff只给出了天数。
DATEDIFF(YEAR,StartDate,EndDate)
DATEDIFF(Month,StartDate,EndDate)
DATEDIFF(Quarter,StartDate,EndDate)
浏览 5提问于2018-09-27得票数 19
回答已采纳
我正在尝试从spark-sql中的SQL语句中获取Integer。
var num_en = ctx.sql("SELECT count(*) FROM table WHERE lang = 'en'")
num = num_en.collect()(0)
num_en是一个SchemaRDD，根据我得到的错误，num是一个“行”。
<console>:144: error: type mismatch;
found : org.apache.spark.sql.Row
(which expands to) org.apache.sp
浏览 0提问于2014-07-03得票数 4
回答已采纳
我使用下面的代码来重新构造一个数据格式，使用它的行来进行整形。
dataframe包含产品更改其ID的日期，但是为了将其与包含事务的巨大的其他数据date连接起来，我需要一个新的列来定义有效的ID范围。
例如，如果产品A更改为产品B，生效日期为01/01，然后更改为产品C生效日期为03/01，则需要同一行中的开始日期和结束日期，这样我就可以根据产品有效的日期B(或C)将其加入到庞大的事务数据过滤中，以便正确地将产品重命名为它们的实际有效ID。
另一条信息，df_MPC大约有800行，不会有更多的增长。
因此，我正在尝试的方法(在开发环境中运行时起作用)是折叠的。
MPC数据文件的摘要版本如下：
浏览 0提问于2018-09-04得票数 2
回答已采纳
3回答
基本上，我想使用SQL语句做一个简单的删除，但是当我执行sql脚本时，它会抛出以下错误：
pyspark.sql.utils.ParseException: U“\n缺失”从‘at’a‘(第2行，pos 23)\n\n== SQL ==\n\n删除a.*从adsquare a\n
下面是我使用的脚本：
sq = SparkSession.builder.config('spark.rpc.message.maxSize','1536').config("spark.sql.shuffle.partitions",str(shuffle_v
浏览 4提问于2017-04-20得票数 9
回答已采纳
我安装了sql server 2008，但无法查看表数据。当我右击表格时，它显示select top 1000 rows，edit top 200 rows。如何查看我的表中的所有行？
浏览 0提问于2009-12-24得票数 5
回答已采纳
我有一个dataframe df和一个列名为setp的列来创建我编写的列表。 setp_list=df.select ('setp').distinct().collect()
setp_array=[row.setp for row in setp_list]
setp_array= str(setp_array)[1:-1] 我想在spark.sql语句中使用它 df1=spark.sql(f"select * from table where setp in ({setp_array})"). 我不确定如何显示列表以查看是如何创建的，但我主要希望将其包含
浏览 12提问于2021-01-10得票数 0
回答已采纳
我不是Spark的专家，也不是底层RDD API的专家。
但是，知道催化剂优化引擎，我希望斯派克会尽量减少内存中的努力。
这就是我的处境:我有两张桌子
TABLE GenericOperation (ID, CommonFields...)
TABLE SpecificOperation (OperationID, SpecificFields...)
它们都很大(大约500米，不是大数据，但在标准应用服务器中作为一个整体内存是不可行的)。
也就是说，假设我必须使用Spark (更大的用例的一部分)检索所有在属于SpecificOperation的字段上匹配某些特定条件的GenericOper
浏览 2提问于2018-06-05得票数 1
我正在编写一个Scala脚本，它从表中读取、转换数据并使用Spark显示结果。我正在使用Spark2.1.1.2和Scala2.11.8。我在脚本中使用了两个dataframe实例(df2在下面的代码中)。由于数据是在调用它们的操作时计算的，而不是在声明它们时计算的，所以我预测这个数据将被计算两次。我认为持久化此数据once将提高性能，认为它只计算一次(持久化时)，而不是两次，如果持久化的话。
但是，与不持久化相比，我的脚本运行时间要长10秒。我不知道这是什么原因。如果有人有主意的话，我会很感激的。
我的提交命令行如下：
spark-submit --class TestQuery --mas
浏览 2提问于2017-10-12得票数 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
回答已采纳
我正在尝试加入两个数据处理程序：
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
"left join“需要在一个数据帧上使用"spark.sql.crossJoin.enabled=true”或调用"persist()“。
SELECT * FROM LHS left join RHS on LHS.R = RHS.R
如何在没有"spark.sql.crossJoin.enabled=true“和持久化数据帧的情况下使”左连接“工作？
下面的例外发生在Spark 2.3.3和2.4.4中。
线程"main“org.apache.spark.sql.AnalysisException中出现异常:检测到逻辑计划OneRowRelati
浏览 0提问于2019-09-07得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
