使用spark sql进行数据分发
Spark SQL是Apache Spark的一个模块，用于处理结构化数据。它提供了一个用于处理结构化数据的编程接口，并支持SQL查询。使用Spark SQL进行数据分析可以帮助用户更高效地处理大规模数据集。
Spark SQL的主要特点包括：
- 高性能：Spark SQL利用Spark的分布式计算能力，可以在大规模集群上并行处理数据，从而实现高性能的数据分析和查询。
- 兼容性：Spark SQL兼容Hive的元数据、查询语法和UDF（用户定义函数），可以无缝迁移现有的Hive应用到Spark平台上。
- 多种数据源支持：Spark SQL支持多种数据源，包括Hive、JSON、Parquet、Avro、ORC等，可以方便地读取和写入不同格式的数据。
- 实时查询：Spark SQL支持流数据处理，可以进行实时查询和分析。
- 内置函数和优化器：Spark SQL提供了丰富的内置函数，可以方便地进行数据转换和计算。同时，它还具有自动优化器，可以对查询进行优化，提高查询性能。
使用Spark SQL进行数据分析的应用场景包括：
- 大数据分析：Spark SQL可以处理大规模的结构化数据，适用于大数据分析和挖掘任务。
- 实时数据处理：Spark SQL支持流数据处理，可以进行实时查询和分析，适用于实时数据处理场景。
- 数据仓库：Spark SQL可以与Hive集成，支持Hive的元数据和查询语法，适用于构建和管理数据仓库。
- 数据可视化：Spark SQL可以与各种数据可视化工具集成，如Tableau、Power BI等，可以方便地进行数据可视化和报表生成。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云Spark服务：提供了基于Spark的大数据分析服务，支持Spark SQL等模块，详情请参考：https://cloud.tencent.com/product/emr
- 腾讯云数据仓库CDW：提供了基于Spark SQL的数据仓库服务，支持大规模数据存储和查询，详情请参考：https://cloud.tencent.com/product/cdw
请注意，以上推荐的腾讯云产品仅供参考，具体选择应根据实际需求和情况进行评估。
相关·内容
1回答
我试图理解Distribute by子句，以及如何在Spark-SQL中使用它来优化Sort-Merge Joins。
根据我的理解，Spark优化器将根据连接键(洗牌阶段)分发两个参与表(连接)的数据集，以便在同一个分区中共同定位相同的键。如果是这样的话，那么如果我们在sql中使用distribute by，那么我们也在做同样的事情。
那么，如何使用distribute by来改善join性能呢？还是在通过加载过程将数据写入磁盘时使用distribute by更好，以便以后使用该数据的查询不必对其进行洗牌，从而从中受益？
请您用一个真实的例子来解释一下如何在Spark中使用distribut
浏览 0提问于2019-08-09得票数 2
1回答
我是spark的新手，正在尝试理解如何在spark中处理倾斜的数据。我已经创建了两个表employee和department。员工对其中一个部门的数据进行了倾斜。 其中一种解决方案是广播部门表，这很好用。但是我想知道如何在下面的代码中使用加盐技术来提高性能。 from pyspark.sql import SparkSession
import pyspark.sql.functions as f
spark = SparkSession.builder.appName("skewTestSpark").config("spark.sql.warehouse.dir&
浏览 24提问于2020-09-06得票数 1
我试图使用cloudera库进行时间序列预测，但无法使用sbt加载库。
下面是build.sbt文件。我可以看到maven回购有0.4.0的disted版本，所以不知道我做错了什么。
有人能帮我知道我对sbt文件做错了什么吗？
import sbt.complete.Parsers._
scalaVersion := "2.11.8"
name := "Forecast Stock Price using Spark TimeSeries library"
val sparkVersion = "1.5.2"
//resolvers ++
浏览 1提问于2018-01-15得票数 0
回答已采纳
我在运行一个Spark应用程序时遇到了一些问题，该应用程序在Spark2.0.0中从Cassandra读取数据。
我的代码工作如下：
DataFrameReader readerCassandra = SparkContextUtil.getInstance().read()
.format("org.apache.spark.sql.cassandra")
.option("spark.cassandra.connection.host", [DATABASE_IP])
浏览 0提问于2018-09-27得票数 0
回答已采纳
我通常使用sbt package编译.sbt文件，其中包含：
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"
libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.1.0"
我对我想要测试的星火源做了一些修改，所以上面的内容不再起作用了，因为它看不到变化。如何使用本地版本的Spark进行编译？
浏览 4提问于2017-07-13得票数 1
回答已采纳
1回答
我有大约80 it的数据，一切都很顺利，直到最后的洗牌任务出现，所有的任务都在30分钟内完成，但是最后的任务需要超过2个小时才能完成。 Joins：(左连接)连接3个表，其中一个表相对较小(2MB)数据，对于那个设置广播变量，即使我删除了第3个表，它也没有解决我的问题。
下面是配置的参数。
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "904857600")
spark.conf.set("spark.cleaner.referenceTracking.blocking",
浏览 8提问于2022-03-02得票数 1
回答已采纳
1回答
Scala异常
、、
我正在学习Scala编程，使用Windows7和最新的Spark2.2.0版本在Apache .I am中编写单词计数驱动程序。在执行程序时遇到下面提到的错误。
如何修正和取得结果？
SBT
命名为:=“:=”版本:= "0.1“scalaVersion：= "2.12.3”val sparkVersion = "2.2.0“libraryDependencies ++= Seq( "org.apache.spark”%“火花-核心2.11”% sparkVersion，"org.apache.spark“%”火花-sql_2.11% sparkVer
浏览 1提问于2017-09-24得票数 0
回答已采纳
当我试图在Cloudera5.5.1版本上使用java.lang.NoSuchFieldError: INT_8执行火花作业时，我得到了OOzie :INT_8错误。如对此有任何帮助，将不胜感激。
请在下面找到错误堆栈。
16/01/28 11:21:17 WARN TaskSetManager: Lost task 0.2 in stage 20.0 (TID 40, Zlab-physrv1): java.lang.NoSuchFieldError: INT_8
at org.apache.spark.sql.execution.datasources.parquet.Cataly
浏览 2提问于2016-01-28得票数 0
回答已采纳
有一个IngiteRDD可以反映底层缓存的变化，这在某些情况下非常好地克服了Spark的RDD的不变性的缺点。
有一个运行ANSI sql (not spark sql)的IgniteRDD.sql方法，该方法返回一个DataFrame，这是spark sql的概念。当我得到这个DataFrame对象时，我可以把它当做一个普通的DataFrame使用吗?我不需要认为它来自Ignite的世界？也就是说，我可以注册临时表，然后与其他DataFrame进行分布式连接。在执行sql时，Ignite是使用Spark SQL引擎还是使用Ignite引擎来运行sql？
浏览 0提问于2016-12-13得票数 0
我很好奇，因为当我们使用spark -e运行查询时，查询在哪里执行。
spark-sql -e "SELECT count(*) FROM table"
而且，当我们计算时，这个动作是否只在驾驶员身上调用？
浏览 3提问于2022-02-11得票数 0
2回答
与JDBC结合使用
、、、、
我在做什么？
我希望使用Flask构建一个API服务，从一个数据库中提取数据，进行一些数据分析，然后将新数据加载到一个单独的DB中。
怎么了？
如果我单独运行Spark，我可以访问数据库，执行分析并加载到db。但是，在Flask应用程序(api路由)中使用它们时，相同的函数不能工作。
我怎么做的？
首先，我启动了火花大师和工人。我可以看到，我有一个工人在localhost:8080下的主人。
export JAVA_HOME=$(/usr/libexec/java_home -v 1.8)
../sbin/start-master.sh
../sbin/start-slave.sh spar
浏览 2提问于2019-01-16得票数 2
回答已采纳
我现在正在学习Spark，它似乎是Pandas Dataframe的大数据解决方案，但是我有一个让我不确定的问题。
目前，我使用HDF5存储比内存更大的Pandas数据文件。HDF5是一个很好的工具，它可以让我在熊猫的数据上分块。因此，当我需要对大型Pandas数据进行处理时，我将以块的形式进行处理。但Pandas不支持分布式处理，HDF5仅适用于单个PC环境。
使用Spark可能是解决方案，但我对Spark的理解是，dataframe必须能够适应内存，并且一旦作为Spark数据存储加载，Spark将将数据分发给不同的工作人员进行分布式处理。
我的理解正确吗？如果是这样的话，那么星火如何处理大
浏览 0提问于2015-10-29得票数 5
回答已采纳
我尝试使用JDBC和spark dataframe从docker中托管的postgres数据库中检索数据。postgres端口在我的Kubernetes集群中作为nodeport打开。 使用以下命令设置连接： val postgres_url = s"$databaseHost:32020"
val postgres_username = "xxxx"
val postgres_db_name = "yyyy"
//Connexion à postgre et récupération du DataFrame de la table
val
浏览 18提问于2019-05-07得票数 2
回答已采纳
广播时，星火可能会因错误org.apache.spark.sql.errors.QueryExecutionErrors#notEnoughMemoryToBuildAndBroadcastTableError而失败(Spark3.2.1)：
为什么BroadcastExchange需要更多的驱动程序内存？广播不是向所有驱动程序发送数据吗？为什么驱动程序内存是瓶颈？
谢谢。
浏览 7提问于2022-10-20得票数 1
回答已采纳
1回答
我使用木星笔记本运行以下流查询使用星火结构化流。
脚本抛出一个StreamingQueryException。我怎么才能解决这个问题？
下面是我使用的程序的版本：
火花: 2.4.3
Scala: 2.11
KAFKA: 2.1.1
我还使用了其他jar文件：
星星之火-spark 0-10_2.11-2.4.3.jar
kafka-clients-2.1.1.jar
以下是星火消费者代码：
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('Test
浏览 6提问于2019-09-24得票数 1
我正在使用以下命令尝试将spark (2.4.4使用Ananaconda 3 Jupyter Notebook)数据帧写入Pyspark中的拼图文件，并得到一个非常奇怪的错误消息，但我无法解决。如果您有任何见解，我将不胜感激。 df.write.mode("overwrite").parquet("test/") 错误信息如下： --------------------------------------------------------------------------
Py4JJavaError
浏览 20提问于2019-12-07得票数 2
回答已采纳
在对数据的某一列执行聚合操作之前，我正在尝试对数据进行预分区。我有3个工作节点，我希望每个分区在我要分区的列中有不重叠的值。我不希望出现两个分区在列中可能具有相同值的情况。
如果我有以下数据
ss_item_sk | ss_quantity
1 | 10.0
1 | 4.0
2 | 3.0
3 | 5.0
4 | 8.0
5 | 13.0
5 | 10.0
然后，以下分区是令人满意的：
分区1
ss_item_sk | ss_quantity
1
浏览 1提问于2019-01-06得票数 0
回答已采纳
我正在运行以下代码(星火版本3.0.1)
case class PubData(publisher_id:Int, country:String, platform:String)
case class PubRes(publisher_id:Int, status:String)
import spark.sqlContext.implicits._
val ds = obSpark.spark.table(tbl)
.select("publisher_id", "country", "platform")
.as[PubData
浏览 6提问于2021-01-13得票数 0
我想在3列上划分一个数据格式"df1“。对于这3列，这个dataframe正好有990个唯一的组合：
In [17]: df1.createOrReplaceTempView("df1_view")
In [18]: spark.sql("select count(*) from (select distinct(col1,col2,col3) from df1_view) as t").show()
+--------+
浏览 1提问于2018-06-05得票数 6
回答已采纳
问:如何在jupyter SparkContext会话中更改spark.sql.pivotMaxValues属性
我做了下面的代码更改以增加spark.sql.pivotMaxValues。可悲的是，在重新启动jupyter并再次运行代码后，它对产生的错误没有任何影响。
from pyspark import SparkConf, SparkContext
from pyspark.mllib.linalg import Vectors
from pyspark.mllib.linalg.distributed import RowMatrix
import numpy as np
try:
浏览 0提问于2018-05-25得票数 3
我用的是火花1.5。
我有两份表格的数据：
scala> libriFirstTable50Plus3DF
res1: org.apache.spark.sql.DataFrame = [basket_id: string, family_id: int]
scala> linkPersonItemLessThan500DF
res2: org.apache.spark.sql.DataFrame = [person_id: int, family_id: int]
libriFirstTable50Plus3DF有766,151记录，linkPersonItemLessThan
浏览 5提问于2016-12-13得票数 71
回答已采纳
从正式文档中可以看到，它首先将表加载到Spark中，然后使用.sql()执行查询。
words = spark.read.format('bigquery') \
.option('table', 'bigquery-public-data:samples.shakespeare') \
.load()
words.createOrReplaceTempView('words')
# Perform word count.
word_count = spark.sql(
'SELECT word, SUM
浏览 3提问于2019-11-19得票数 1
回答已采纳
我已经在我的spark streaming应用程序中启用了以下列出的配置，但我无法推断设置这些参数后的性能优势，如果你们中的任何一个知道任何方法来验证矢量化是否按预期/正确启用！
注意:我正在使用Spark 2.3，并将我的应用程序中的所有数据转换为原生orc格式1.4版本。
sparkSqlCtx.setConf("spark.sql.orc.filterPushdown", "true")
sparkSqlCtx.setConf("spark.sql.orc.enabled", "true")
sparkSqlCtx.se
浏览 3提问于2018-05-23得票数 1
2回答
我一直在使用dataset Movielens (2000万条记录)，并且一直在使用中的Spark。
我的环境是VirtualBox上的Ubuntu14.4。我有一个主节点和两个从节点。我使用了已发布的Apache、Apache、Scala、sbt。代码是用Scala编写的。
如何将代码和数据集分发到工作节点上？
import java.lang.Math._
import org.apache.spark.ml.recommendation.ALS
import org.apache.spark.ml.recommendation.ALS.Rating
import org.ap
浏览 0提问于2019-06-29得票数 1
回答已采纳
1回答
在下面的示例中，我误解了PySpark的性能。
我有几个DataFrame，因此我加入了它们。
print"users_data"
print users_data.show()
print"calc"
print calc.show()
print"users_cat_data"
print users_cat_data.show()
data1 = calc.join(users_data, ['category_pk','item_pk'], 'leftouter')
print "
浏览 0提问于2018-01-15得票数 2
回答已采纳
我在spark工作了6个多月。我见过来自数据仓库和SQL背景的人正在使用SQL实现聚合和其他转换逻辑
spark.sql()
(where spark is the sparkSession object)
直接通过配置单元表或在将数据帧注册为TempView之后使用
dataframe.createOrReplaceTempView().
但是，如果我们看到，我们还有其他选择，比如windows函数或其他选项，它们可以直接在dataframes.Or上实现，甚至我们可以将函数注册为UDF，并可以在dataframe上实现。
假设我需要在数据帧CITY_CENSUS上实现按城市分组的人口
浏览 1提问于2019-05-05得票数 0
下面是我的代码：
val opts = Map("allowUnquotedFieldNames" -> "true")
val df_withSchema = df.withColumn("Data", from_json(col("Item.Data.S"),schema, opts))
以下是错误：
(<console>:198: error: overloaded method value from_json with alternatives:
(e: org.apache.spark.s
浏览 7提问于2022-03-03得票数 0
回答已采纳
Databricks的深度学习管道是一个带有Python的星火包，其目的是使来自Tensorflow/Keras的深度学习模型能够在Spark上运行并以DataFrame作为输入。
这听起来很酷，因为这样可以在分布式数据集上以分布式方式运行图像识别(特别是)，从而确保可伸缩性。
然而，在阅读sparkdl包的代码( 的深度学习管道的Python )时，我观察到了一些让我怀疑整个事情的可伸缩性的东西。
首先，sparkdl使用图像文件夹(例如需要分类)作为输入，并将其加载到星火DataFrame中。
然后，为了在这些图像上建立一个深入的学习模型，sparkdl的“拟合”方法做了以下工作。
它
浏览 1提问于2017-09-28得票数 1
回答已采纳
我正在尝试使用spark摄取一个JSON文件。我正在手动应用模式来创建Dataframe。问题是，即使模式的单个记录不匹配，它也会将整个文件(所有记录)移动到损坏的列中？
数据
[{
"RecordNumber": 2,
"Zipcode": 704,
"ZipCodeType": "STANDARD",
"City": "PASEO COSTA DEL SUR",
"State": "PR"
},
{
"RecordNumber
浏览 2提问于2020-10-22得票数 0
我正在实现基于对文本进行归一化的情感分析(在调用udf之前，我正在按照柠檬化函数的要求将格式转换为字符串)，但我面临一个我无法理解的酸洗错误。
PicklingError:无法序列化对象: TypeError:无法序列化“_io.BufferedReader”对象
下面是我的代码
def get_part_of_speech(word):
probable_part_of_speech = wordnet.synsets(word)
pos_counts = Counter()
pos_counts["n"] = len( [ it
浏览 2提问于2020-08-04得票数 1
1回答
我是星火世界的新手，在一些概念上苦苦挣扎。
当使用由Kafka提供的Spark结构化流源时，并行性是如何发生的？
让我们考虑下面的代码片段：
SparkSession spark = SparkSession
.builder()
.appName("myApp")
.getOrCreate();
Dataset<VideoEventData> ds = spark
.readStream()
.format("kafka")
...
gDataset = ds.gro
浏览 3提问于2018-01-13得票数 8
回答已采纳
1回答
我有一个试图用pyspark管理的数据集，我想为它选择一个子集。让我说一下，我对spark、scala和pyspark一般都很陌生。
这就是我要做的。
from pyspark.sql import SparkSession
from pyspark.sql.types import *
import pyspark.sql.functions as fs
spark = SparkSession.builder\
.master("local")\
.enableHiveSupport()\
浏览 2提问于2020-02-03得票数 0
我试图使用spark-sklearn库在星火库集群上执行网格搜索。出于这个原因，我在我的nohup ./spark_python_shell.sh > output.log & shell上运行bash以点燃Spark集群，并且运行我的python脚本(参见下面的spark-submit \ --master yarn 'rforest_grid_search.py')：
SPARK_HOME=/u/users/******/spark-2.3.0 \
Q_CORE_LOC=/u/users/******/****** \
ENV=loca
浏览 2提问于2018-10-25得票数 2
回答已采纳
我正在尝试从avro读取数据，按字段重新划分数据并将其保存为avro格式。下面是我的示例代码。在调试过程中，我不能在我的数据帧上显示(10)。它会失败，并显示以下错误。有没有人能帮我弄明白我在代码行中做错了什么？
代码：
import org.apache.spark.sql.avro._
val df = spark.read.format("avro").load("s3://test-bucekt/source.avro")
df.show(10)
df.write.partitionBy("partitioning_column").
浏览 40提问于2019-11-27得票数 0
我正在尝试使用JDBC从数据库表中读取一些内容：
val df = spark.read.jdbc("<database url>", "<some table name>", <some DbProperties>)
然后将其写入另一个数据库：
df.write.mode(SaveMode.Append).jdbc("<other database url>", "<same table name>", <some DbProperties>)
如果我们不在
浏览 0提问于2020-07-09得票数 1
我正在运行一个简单的sparkSQL查询，它在两个数据集上进行匹配，每个数据集大约是500 is。所以整个数据都在1TB左右。
val adreqPerDeviceid = sqlContext.sql("select count(Distinct a.DeviceId) as MatchCount from adreqdata1 a inner join adreqdata2 b ON a.DeviceId=b.DeviceId ")
adreqPerDeviceid.cache()
adreqPerDeviceid.show()
作业工作良好，直到数据加载(10k任务分配
浏览 3提问于2016-10-17得票数 1
回答已采纳
2回答
在一个相当新的环境中深入研究spark的POC，并检查spark功能，但在pyspark终端中运行sql查询时出现问题，而Hive正在工作，因为我们可以查询元数据。 你知道这里发生了什么以及如何解决这个问题吗？ $ pyspark --driver-class-path /etc/spark2/conf:/etc/hive/conf
>>> from pyspark.sql import SparkSession
>>> from pyspark.sql import Row
>>> spark = SparkSession \
...
浏览 500提问于2020-08-19得票数 2
我尝试通过创建HiveContext来通过Spark 2.2.1查询Hive表。事实证明，Spark (无论我是通过spark-submit提交我的作业，还是在pyspark shell中运行它-同样的效果)是有效的，但只能在Hive中看到默认数据库，而不能看到任何其他数据库。这个问题似乎已经存在一段时间了，所有的建议都是关于调整Spark参数，比如--deploy-mode和--master，并显式地将hive-site.xml文件传递给Spark。
在阅读了有关此问题的所有内容后，我将spark-submit命令更改为以下命令：
/bin/spark-submit --driver-cla
浏览 3提问于2018-02-23得票数 2
为了将大量SQL数据加载到Spark & ML中，下面哪个选项的性能更好。
选项1:使用Spark连接器将SQLData直接加载到Spark。
选项2:使用Sqoop以csv格式将SQLData加载到HDFS，然后使用Spark从HDFS读取数据。
请建议以上哪一种方法可以很好地将大型SQL数据加载到Spark。
浏览 12提问于2015-11-18得票数 3
回答已采纳
1回答
我有这样的数据
+-------+------------------------+
|key | data|
+-------+------------------------+
| 61|[a -> b, c -> d, e -> f]|
| 71|[a -> 1, c -> d, e -> f]|
| 81|[c -> d, e -> f] |
| 91|[x -> b, y -> d, e -> f]|
| 11|[a -&
浏览 1提问于2020-04-09得票数 1
回答已采纳
1回答
我需要将数据集数据划分为6列: region/year/month/day/id/quadkey，其中在顶层我只有二进制区域状态，而在最底层实际上是它进入许多分区的位置。假设我们有两个区域/通常是一年/通常是一个月/3-4天/100-150个I/ 50-200个四键，当我执行这个操作时，我得到了非常不平衡的随机操作，有时执行器会因为超出内存限制而失败。此外，我从History UI中注意到，hat阶段的一些任务非常大(约15 1Gb)，而其他任务则小得多(约1 1Gb)。
我试着去玩
sqlContext.setConf("spark.sql.shuffle.partitions
浏览 0提问于2019-04-11得票数 1
我有一份胶水工作：
从几个数据目录创建动态框架
更改为触发数据格式。
连接4个数据文件并完成聚合。
用csv/parquet文件类型写到s3。
它对中型数据源(约20G数据)、G1x DPU、执行时间40分钟的20名工作人员没有问题。但是，当数据量增加到60G时，如果使用G2x DPU和50名工作人员，则耗时将增加到4-6小时，从而导致错误：
21/08/07 16:41:27 ERROR ProcessLauncher: Error from Python:Traceback (most recent call last):
File "/tmp/test
浏览 4提问于2021-08-07得票数 1
我试图从Python中查询snappydata，一些答案在StackOverflow中说Python不能连接到远程spark集群。谁能帮助我如何连接到snappydata集群，并让一个简单的查询工作？ 我正在尝试的代码- from pyspark.sql.snappy import SnappySession
snappy = SnappySession.builder.appName("test") \
.master("local[*]") \
浏览 19提问于2020-07-18得票数 0
我的以下代码是使用spark从hive表中读取数据。该表中有1亿条记录。当我在Rdd中选择这么多记录并尝试执行result.show()时，它给出了严重的问题异常。
我基本上是想通过从这个表中选择几列来插入其他表中的记录，以获得1亿条记录集。
下面是我的代码：
import org.apache.spark.sql.functions._
import org.apache.spark.sql._
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val hiveContext = new org.apache.spark.sql
浏览 0提问于2017-11-07得票数 1
我在集群模式下运行spark，并通过JDBC从RDBMS读取数据。
根据火花，这些分区参数描述了如何在从多个工作人员并行读取表时对表进行分区：
partitionColumn
lowerBound
upperBound
numPartitions
这些是可选的参数。
如果我不具体说明会发生什么情况：
只有一名工人读取全部数据？
如果它仍然读取并行，它如何划分数据？
浏览 3提问于2017-03-31得票数 15
回答已采纳
我正在尝试迭代我从使用Spark执行结构化查询时获得的Row(吡火花)。代码大致如下所示：
spark = SparkSession \
.builder \
.appName("Sessions")\
.config(...) \
.getOrCreate()
stuff = spark.read.format("parquet").options(mergeSchema=False).load(location)
result_set = spark.sql("""sql without udfs
浏览 0提问于2018-02-27得票数 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
回答已采纳
3回答
在spark (如 )中，有许多方法来验证两个数据帧的架构。但是我只想在SQL中验证两个数据帧的模式，我指的是SparkSQL。
示例查询1:
SELECT DISTINCT target_person FROM INFORMATION_SCHEMA.COLUMNS WHERE COLUMN_NAME IN ('columnA','ColumnB') AND TABLE_SCHEMA='ad_facebook'
示例查询2:
SELECT count(*) FROM information_schema.columns WHERE table_n
浏览 6提问于2018-09-04得票数 1
我有一个数据，并写入到S3桶目标位置。在代码中，合并用于加载数据和获取SparkOutOfMemoryError.The当前代码，这些代码使用了多个项目，并且看到了许多推荐的重新分区解决方案，它对我起了作用。即使它没有记录，合并也不起作用。有没有其他的方法来解决这个问题而不改变为重新划分？
代码：
empsql = 'Select * From Employee'
df = spark.sql(empsql) ##Spark is configured
df.coalesce(2).write.mode('overwrite').format("parq
浏览 8提问于2020-06-11得票数 2
我想在RDD上的转换中访问伴生对象的方法。为什么以下选项不起作用：
import org.apache.spark.rdd.RDD
import spark.implicits._
import org.apache.spark.sql.{Encoder, Encoders}
class Abc {
def transform(x: RDD[Int]): RDD[Double] = { x.map(Abc.fn) }
}
object Abc {
def fn(x: Int): Double = { x.toDouble }
}
implicit def abcEncoder
浏览 8提问于2017-05-07得票数 4
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
