参数化spark partition by子句
是指在Spark中使用参数来指定分区的依据。Spark是一个开源的分布式计算框架，可以用于大规模数据处理和分析。分区是将数据集划分为更小的部分，以便在集群中并行处理。
在Spark中，partition by子句用于指定数据分区的依据。通过将数据集按照指定的列进行分区，可以提高数据处理的效率和性能。参数化partition by子句意味着可以使用变量或参数来动态指定分区的列，而不是固定的列名。
参数化spark partition by子句的优势在于灵活性和可重用性。通过使用参数，可以根据不同的需求和场景来动态指定分区的列，而不需要修改代码。这样可以提高代码的可维护性和可扩展性。
参数化spark partition by子句的应用场景包括但不限于以下几个方面：
- 数据仓库和数据湖：在构建数据仓库和数据湖时，可以使用参数化partition by子句来按照不同的维度对数据进行分区，以便更好地支持数据查询和分析。
- 数据清洗和转换：在数据清洗和转换过程中，可以使用参数化partition by子句来按照不同的规则对数据进行分区，以便更好地处理和转换数据。
- 数据备份和恢复：在数据备份和恢复过程中，可以使用参数化partition by子句来按照不同的时间戳或其他标识对数据进行分区，以便更好地管理和恢复数据。
腾讯云提供了一系列与Spark相关的产品和服务，包括云上Elastic MapReduce（EMR）和云原生数据仓库（CDW）。EMR是一种大数据处理服务，可以在云上快速部署和管理Spark集群。CDW是一种基于Spark的云原生数据仓库，可以提供高性能和弹性的数据存储和分析能力。
更多关于腾讯云Spark相关产品和服务的信息，可以参考以下链接：
- 腾讯云Elastic MapReduce（EMR）：https://cloud.tencent.com/product/emr
- 腾讯云云原生数据仓库（CDW）：https://cloud.tencent.com/product/cdw
相关·内容
1回答
我试图使用Foreach分区来迭代一个数据文件，以便将一个值插入到数据库中。我使用了前端分区，对行进行分组，并使用foreach迭代每一行。请在下面找到我的密码，
val endDF=spark.read.parquet(path).select("pc").filter(col("pc").isNotNull);
endDF.foreachpartition((partition: Iterator[Row]) =>
class.forname(driver)
val con=DriverManager.connection(jdbcu
浏览 0提问于2021-12-02得票数 0
1回答
我正在运行下面的sql，其中表中的数据围绕着4,34,836,959记录。需要3分钟以上才能得到结果。
select distinct col_1,col_2,col_3,col_4,col_5,
to_date(concat(year(col_6),'-',month(col_6), '-1')) as col_6_new,col_7,
cast(first_value(col_11) over (partition by col_1,col_2,col_5,col_4,concat(year(col_6),'-',month(col_6)
浏览 3提问于2020-07-23得票数 1
我有一个spark应用程序，我不能在Yarn上启动。我得到了空指针异常。(在主本地应用程序上工作正常)。我的应用程序正在将嵌套的XML转换为JSON。
我使用：val xml_rdd = spark.sparkContext.textFile(path)将XML加载到RDD中
我正在使用scala.xml.XML.loadString(xmlStr).toJson.toString()将XML转换为JSON。toJson方法需要隐式编写器，所以我正在创建它：
implicit val writer: JsonWriter[Elem] = new JsonWriter[Elem] {
浏览 24提问于2020-05-28得票数 0
3回答
我通过Spark使用hive。在我的spark代码中，我有一个Insert Insert my table查询。输入数据为200+gb格式。当Spark写入分区表时，它会生成非常小的文件(以kb为单位的文件)。因此，现在输出分区表文件夹中有5000+小kb文件。我想把这些合并成几个大的MB文件，可能是几个200mb的文件。我尝试使用配置单元合并设置，但它们似乎不起作用。
'val result7A = hiveContext.sql("set hive.exec.dynamic.partition=true")
val result7B = hiveContext.
浏览 6提问于2015-06-24得票数 8
回答已采纳
我在azure data-bricks上运行spark。我的要求是，我需要从oracle db中提取数据并将其输出到文件中。 Spark verison - 2.4
Databricks cluster size - 8 nodes,8 cores for each node. 因此，为了获得更多的并行性，我在oracle查询中对分区列使用了哈希算法。 example - mod(ora_hash(id), 64) as partition_key 问题是，尽管我在Data-bricks集群中有64个核心可用，但只有8个核心在运行以从oracle中提取数据。 请找到附件中的屏幕截图以供参考。
浏览 53提问于2019-04-17得票数 0
我在Spark2.3.1中SparkSQL的最后一个聚合中遇到了一个非常麻烦的问题。它似乎在500,000个(逻辑SQL，而不是Spark)分区中产生了大约4个坏结果--即，不是按指定的分区和顺序最后显示的值，大约是50 me的记录。较小的批处理更糟糕--每个批处理的错误数量似乎相当一致，尽管我不认为我尝试过小于100,000个逻辑SQL分区的任何分区。
我有大约66个第一次或最后一次聚合、一个复合(日期、整数)逻辑sql分区键和一个复合(字符串、字符串)排序键。我尝试将四个字符的数值转换为整数，然后将它们组合成一个整数。这些举措都没有解决问题。即使只有一个整数排序关键字，我也会得到一些不好的
浏览 2提问于2019-11-19得票数 1
在spark SQL中，我尝试仅当列的前一行不为null时才获取该列的前一行的值。由于忽略空值不可用，因此替代方案将是排名并获得该值的最大值。
lag( val ignore nulls) over ()
SELECT id, val, val_partition, MAX(val) over (partition by val_partition)
FROM (
SELECT
id,
val,
sum(case when val is null then 0 else 1 end) over (order
浏览 93提问于2020-02-07得票数 1
回答已采纳
我成功地集成了kafka和spark。我想从kafka发送流到spark.and，现在我可以发送流到spark.I了。我想要RDD中的流，所以我使用createRDD()函数来创建RDD。但我只从rdd的kafka那里得到了一些按摩师。因为它依赖于偏移量，所以请任何人告诉我如何在kafka- range.So CreateRDD()函数中设置offsetRange()。
浏览 0提问于2016-10-26得票数 0
我有一个关于spark的问题。如果我在spark驱动程序上定义了到远程服务器的长期套接字连接，我还可以使用该套接字与spark worker.How上的远程服务器通信。
例如，我在spark driver上定义了一个redis客户端，我也可以通过spakr worker上的redis客户端来读写数据。
val redisClient = Jedis.RedisClient(host, port)
rdd1.map(x => {
val value = redisClient.get(x)
value
})
也许redis并不是一个完全正确的例子。我想知道spark是否可以将我的长
浏览 16提问于2019-08-03得票数 0
我正试图根据以下数据框架- salesDF的收入对列“产品”进行排名。
salesDF=
+-------------+-------+---------+----------+-------+
|transactionID|Product| category|produtType|Revenue|
+-------------+-------+---------+----------+-------+
| 105| Lenova| laptop| high| 40000|
| 111| Lenova| tablet| medium
浏览 3提问于2020-05-03得票数 1
回答已采纳
我正在研究在kafka中存储kafka偏移量用于Spark结构化流媒体，就像它对DStreams stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)一样，与我正在寻找的相同，但用于结构化流媒体。它是否支持结构化流媒体？如果是，我如何实现它？
我知道使用.option("checkpointLocation", checkpointLocation)的hdfs检查点，但我对内置的偏移量管理很感兴趣。
我期望kafka只在内部存储偏移量，而不使用spark hdfs检查点。
浏览 10提问于2019-05-16得票数 3
1回答
我试图将星火DataFrame保存到带有.saveAsTable() in pySpark的Hive表(Parquet)中，但是继续运行到内存问题，如下所示：
org.apache.hadoop.hive.ql.metadata.HiveException: parquet.hadoop.MemoryManager$1:
New Memory allocation 1034931 bytes is smaller than the minimum allocation size of 1048576 bytes.
第一个数字(1034931)通常在不同的运行中不断变化。我知道第二个数字(104
浏览 1提问于2017-05-17得票数 4
回答已采纳
我正在尝试运行下面的示例，尝试从星火DataFrame创建一个Hive表。当我用master=local调用submit时，代码可以工作，但是当我用master=yarn调用它时，它会抛出异常。下面是调用:星星之交--类test.sandbox.HDPRiskFactor --主纱--命名为“危险因素”。/hdrisk-0.0.1-SNAPSHOT.jar--此外，我还从Hive控制台创建了一个名为"default.geolocation“的表，但当我调用show()时，我无法从spark中看到它。我尝试将Yarn模式下的executor计数设置为0，但这也不起作用。1)为什么代码与本
浏览 2提问于2019-02-27得票数 0
回答已采纳
我有一个BQ表，它由缺省的_PARTITIONTIME分区。我想把它的一个分区读到Spark dataframe (PySpark)。但是，spark.read应用编程接口似乎不能识别分区列。下面是代码(不起作用)：
table = 'myProject.myDataset.table'
df = spark.read.format('bigquery').option('table', table).load()
df_pt = df.filter("_PARTITIONTIME = TIMESTAMP('2019-01-3
浏览 0提问于2019-07-31得票数 1
1回答
如何-分散作业，以免耗尽内存
、、、、
我试图运行一些火花作业，但通常执行者会耗尽内存：
17/02/06 19:12:02 WARN TaskSetManager: Lost task 10.0 in stage 476.3 (TID 133250, 10.0.0.10): ExecutorLostFailure (executor 12 exited caused by one of the running tasks) Reason: Container marked as failed: container_1486378087852_0006_01_000019 on host: 10.0.0.10. Exit statu
浏览 9提问于2017-02-06得票数 2
回答已采纳
3回答
我试图通过酸洗来序列化Spark RDD，并将酸洗过的文件直接读取到Python中。
a = sc.parallelize(['1','2','3','4','5'])
a.saveAsPickleFile('test_pkl')
然后，我将test_pkl文件复制到本地。如何将它们直接读入Python？当我尝试普通的pickle包时，当我尝试读取‘test_pkl’的第一个pickle部分时，它失败了：
pickle.load(open('part-00000','rb&#
浏览 0提问于2015-11-19得票数 4
2回答
在spark-shell中，如何加载现有的Hive表，但只加载其中的一个分区？
val df = spark.read.format("orc").load("mytable")
我正在寻找一种方法，所以它只加载这个表的一个特定分区。
谢谢!
浏览 19提问于2020-04-30得票数 0
回答已采纳
1回答
我有一个相当重的金属库，它是通过Java/JNI连接器访问的，我想在集群中通过Spark使用它。但是，在第一次使用之前需要初始化库，这需要大约30秒的时间。所以我的问题是，Spark是否有某种机制在作业开始时预初始化这些库，以便在实际使用中不需要初始化开销？
浏览 0提问于2017-02-10得票数 3
我的代码在emr中不起作用
finalDF.createOrReplaceGlobalTempView("temp_visits")
spark.sql(s"insert overwrite table test PARTITION (date) SELECT * from temp_visits")
我得到以下异常
Exception in thread "main" org.apache.spark.sql.AnalysisException: Table or view not found: temp_visits; line 1 pos
浏览 1提问于2018-07-19得票数 4
回答已采纳
我正在尝试把我的桌子在PostgreSQL上装载到火花上。我已经成功地使用jdbc从PostgreSQL读取了这个表。我有一个用R写的代码，我想在表上使用它，但是我不能访问R中的数据。
使用以下代码连接
val pgDF_table = spark.read
.format("jdbc")
.option("driver", "org.postgresql.Driver")
.option(
浏览 0提问于2019-07-29得票数 0
回答已采纳
我需要为spark应用程序中的每个spark executor分配一个唯一的整数id。我需要从在executor上运行的任务中检索executor id。执行器id将与其它数据元素(时间戳、mac地址等)一起用于生成唯一的64位密钥。如何将唯一的整数键分配给Apache Spark Java应用程序中的每个Apache Spark Executor？
浏览 27提问于2018-01-18得票数 0
回答已采纳
我试图用spark.catalog.createTable创建一个表。它需要一个名为"id“的分区列。
基于Scala中的，我尝试：
df = spark.range(10).withColumn("foo", F.lit("bar"))
spark.catalog.createTable("default.test_partition", schema=df.schema, **{"partitionColumnNames":"id"})
但这是行不通的。它使用以下属性在Hive中创建一个表：
CREA
浏览 3提问于2021-09-28得票数 2
回答已采纳
当我使用Python在AWS Glue中运行Glue作业样板时，我会得到一个错误
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
glueContext = GlueContext(SparkContext.getOrCreate())
spark
浏览 0提问于2018-09-13得票数 5
回答已采纳
即使使用.cache() data，Spark似乎仍然序列化每个任务运行的数据。请考虑以下代码：
class LoggingSerializable() extends Externalizable {
override def writeExternal(out: ObjectOutput): Unit = {
println("xxx serializing")
}
override def readExternal(in: ObjectInput): Unit = {
println("xxx deserializing")
浏览 0提问于2020-07-29得票数 2
我正在尝试用Spark将一个大的分区数据集写入磁盘，而partitionBy算法正在努力解决我尝试过的两种方法。
分区是严重倾斜的-一些分区是巨大的，另一些是很小的。
问题#1
当我在partitionBy之前使用重新分区时，斯派克将所有分区写成一个文件，即使是巨大的
val df = spark.read.parquet("some_data_lake")
df
.repartition('some_col).write.partitionBy("some_col")
.parquet("partitioned_lake")
这
浏览 1提问于2018-10-28得票数 23
回答已采纳
4回答
我正在使用(Spark2.2.0)。
我试图按以下方式删除Hive分区：
spark.sql(""ALTER TABLE backup DROP PARTITION (date < '20180910')"
得到了以下例外：
org.apache.spark.sql.catalyst.parser.ParseException:不匹配的输入'<‘期待{’'，‘，’}(第1行，pos 42)
我知道这是一个应该在我的版本中修复的未决问题，但我仍然有例外。
我有什么办法把分区从星火中扔掉呢？还有另一个实现可以这样做吗？
谢
浏览 5提问于2018-09-27得票数 7
1回答
目前，我有一个ETL作业，它读取几个表，执行某些转换并将它们写回每日表。
我在sql 中使用了以下查询：“Insertintodbname.tablename分区(Year_month) SELECT * from Spark_temp_table "中插入所有这些记录的目标表在一年X月级别上进行分区。每天生成的记录并不多，因此我要在X年的月份级别上进行分区。
但是，当我检查分区时，我的代码每天运行(代码必须每天运行)，它有小到50 my的文件，最终，我的分区中将有大约30个文件，总数约为1500 my。
我想知道是否有办法在一个分区中创建一个文件(或者按块大小限制创建2个-3个文件)
浏览 1提问于2018-10-20得票数 1
我正在尝试使用1.2版本的新连接功能，但是在repl中的repartitionByCassandraReplica函数出现了一个错误。
我试图复制这个网站的例子，并创建了一个带有两个元素的卡桑德拉表(shopping_history)：
import com.datastax.spark.connector.rdd._
import com.datastax.spark.connector.cql.CassandraConnector
import com.datastax.spark.connector._
import com.datastax.driver.core._
case cl
浏览 21提问于2015-04-17得票数 0
1回答
我正在运行一个流式应用程序和处理数据从卡夫卡到卡夫卡使用火花。如果我使用的是最新的，那么它的工作，如预期和运行，没有任何问题。
但是在源方面，我们已经完成了批量事务(20万)，并且使用最早的，然后处理数据。在这种情况下，我们的火花作业不是处理数据，它在三个阶段后被卡住了。有人能建议我如何处理这个问题吗?这样我就可以处理这些海量数据了。
我使用以下配置：
TRIGGERFREQUENCY 1 seconds
STARTINGOFFSETS earliest
--num-executors 6
--driver-cores 6
--driver-memory 8G
--e
浏览 7提问于2022-05-17得票数 0
我正在为我的spark结构化流媒体应用程序构建监控，并且需要获得spark应用程序消耗的某个主题的消费者滞后。我相信spark驱动程序一定知道这个延迟，因为它有执行器的所有元数据。我看不到任何方法可以从任何现有的spark文档或资源中获得这个指标。我检查了streaminQueryListener接口，但它的功能也很有限，因为我们只能从它获得每个查询的指标。
浏览 0提问于2020-03-27得票数 2
1回答
我正在学习Spark，并试图建立一个简单的流媒体服务。
例如，我有一个Kafka队列和一个像这样的Spark作业。该示例使用无状态模式。我想要累积字数，这样如果test在不同的消息中被发送了几次，我就可以得到它所有出现的总数量。
使用等其他示例，我尝试修改我的Kafka流媒体服务
val sc = new SparkContext(sparkConf)
val ssc = new StreamingContext(sc, Seconds(2))
ssc.checkpoint("/tmp/data")
// Create direct kafka stream with br
浏览 3提问于2017-03-27得票数 3
我正在尝试运行一个spark sql语句，并试图在执行聚合的同时执行一个简单的group by；它抱怨说，它无法在我在模式中提供的给定输入列中找到列month，但在教程之后，他们能够运行给定的代码。 代码： StructField[] fields = new StructField[]{
new StructField("level", DataTypes.StringType, false, Metadata.empty()),
new StructField("datetime", DataType
浏览 16提问于2020-08-23得票数 1
回答已采纳
1回答
我正在尝试对示例数据执行简单的Spark转换mapPartition()。但是在这个过程中，我得到了java.io.NotSerializableException: org.apache.spark.InterruptibleIterator异常。
这是我的例外：
java.io.NotSerializableException: org.apache.spark.InterruptibleIterator
Serialization stack:
- object not serializable (class: org.apache.spark.InterruptibleIter
浏览 3提问于2016-12-06得票数 2
3回答
是否可以在覆盖子句中使用where子句，如下所示？
SELECT SUM(amount) OVER(partition by prod_name WHERE dateval > dateval_13week)
我不能使用之前和后面的内部over子句，因为我的日期不在顺序中。所有我需要获取的是小于当前记录的13week日期值的记录。
EDIT :
sum(CASE WHEN dateval >= dateval_13week and dateval <=current_row_dateval then amount else 0 end) over (partition by
浏览 1提问于2014-03-03得票数 10
1回答
我使用Apache将非分区表写入AWS S3，并将其同步到单元格。这是正在使用的DataSourceWriteOptions。
val hudiOptions: Map[String, String] = Map[String, String](
DataSourceWriteOptions.TABLE_TYPE_OPT_KEY -> "MERGE_ON_READ",
DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY -> "PERSON_ID",
DataSourceW
浏览 3提问于2020-10-21得票数 3
2回答
我们正在开发Spark SQL。我们正在使用一些可以为空的字符串字段进行排名。 问题是:在Spark SQL中，null值排在第一位。但是，我们希望null的价值最终出现。因此，我们应用了一个案例，当逻辑。因为我们有Unicode数据，"ZZZZZZZZ“最终不会出现。它将领先于日语和中文地址线。 请告诉我们，使用哪个字符串常量字符串值为空，以便它们在ORDER BY的情况下出现在末尾。 我已经在下面放了示例代码。 SELECT CompanyName,
ROW_NUMBER() OVER
(
浏览 36提问于2019-01-22得票数 1
回答已采纳
我正在sparklyr中处理一个集群问题。训练集中的许多变量是在不同的尺度上测量的，因此按数量级的不同而不同。根据最佳实践，我正试图对数据进行缩放和中心化。
有许多不同的公式可以这样做，最传统的存在(X -)/σ，其中X是随机变量，σ=均值和σ=标准差。我倾向于使用( X= )/ (x_max - x_min)，其中X=随机变量、x=sample均值、x_max=最大值和x_min =最小值。
在使用dplyr应用此转换之后，我将得到一个有用的结果。考虑一下这个例子：
#connect to spark
library(sparklyr)
library(SparkR)
浏览 1提问于2017-12-13得票数 3
回答已采纳
2回答
我正在尝试使用不同的时间戳对hive表进行分区。我有一个带有时间戳的表，但是当我执行hive分区查询时，它告诉我它不是有效的分区列。下面是表格：
+---+-----------------------+
|id |rc_timestamp |
+---+-----------------------+
|1 |2017-06-12 17:18:39.824|
|2 |2018-06-12 17:18:39.824|
|3 |2019-06-12 17:18:39.824|
+---+-----------------------+
spark.sql("SE
浏览 0提问于2019-06-13得票数 3
2回答
我已经看到了一个类似于的问题
但是，我仍然想知道来自特定分区的流数据是否不可能实现？我使用了卡夫卡消费策略在星火流订阅方法。
ConsumerStrategies.Subscribe
这是我尝试订阅主题和分区的代码片段，
val topics = Array("cdc-classic")
val topic="cdc-classic"
val partition=2;
val offsets=
Map(new TopicPartition(topic, partition) -> 2L)//I am not clear with this lin
浏览 0提问于2018-06-07得票数 0
回答已采纳
由于输入数据的数量很大，我设置了spark (spark.sql.shuffle.partitions=1000)的大洗牌分区。但是，输出文件很小(~1GB)，但是它会创建大量的小文件(3000个文件，每个文件小于1Mb)。如何将这些小文件组合成一个大文件？
另一个问题是，为什么输出文件的数量是洗牌分区数量的3倍？
浏览 24提问于2021-12-14得票数 1
我尝试了一下kafka火花结构化流，但是得到了一些异常，比如线程"main“中的异常:无法解析给定的输入列:值、偏移量、分区、键、时间戳、timestampType、主题；。
附加我的代码
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types._
import org.apache.spark.sql.streaming.ProcessingTi
浏览 0提问于2017-07-24得票数 0
我有一个Python函数，它返回一个Pandas DataFrame。我在Spark2.2.0中使用pyspark的调用这个函数。但是我不能将mapPartitions()返回的RDD转换为Spark DataFrame。Pandas会生成此错误：
ValueError: The truth value of a DataFrame is ambiguous. Use a.empty, a.bool(), a.item(), a.any() or a.all().
说明问题的简单代码：
import pandas as pd
def func(data):
pdf = pd.Data
浏览 11提问于2018-01-31得票数 3
回答已采纳
当前上下文:分页查询结果
我需要知道是否有某种方法可以在我的查询中获得计算值，比如row_number，它计算特定数量的行上的下一个新值。
如果我使用row_number，它将在我结果中的identity列中生成一种类型，无论我是否希望在指定的行数上对其进行合并。如下所示：
每页行数:3
RowNum, PageNum, Value
1, 1, rowa
2, 1, rowb
3, 1, rowc
4, 2, rowd
5, 2, rowe
6, 2, rowf
7, 3, rowg
浏览 0提问于2013-03-19得票数 0
回答已采纳
1回答
我已经安装了DCOS集群，在安装spark时，我提供了hdfs-site.xml和core-site.xml外部集群的详细信息，我试图使用spark访问这些集群。配置了/etc/host文件的DCOS集群的HDFS详细信息，我正在尝试运行我的火花作业在帮助下DCOS命令
dcos火花运行-提交-args=‘-dspak.mesos.粗=真-驱动程序-核心1-驱动程序-内存1024M -文件 -类SparkAvroConsumer ’
火花作业执行遇到异常。此外，spark运行的代理正在尝试访问本地主机，而不是HDFS集群。是否有任何其他配置，我需要访问外部HDFS？
java.net.Conn
浏览 0提问于2018-02-22得票数 0
在Spark中读取ORC文件时，如果在路径中指定分区列，则该列将不会包含在数据集中。例如，如果我们有
val dfWithColumn = spark.read.orc("/some/path")
val dfWithoutColumn = spark.read.orc("/some/path/region_partition=1")
那么dfWithColumn将有一个region_partition列，但dfWithoutColumn没有。我如何指定要包含所有列，即使它们已分区？
我在scala上使用的是spark 2.2。
EDIT:这是一个可重用的S
浏览 1提问于2018-09-13得票数 5
1回答
由于以下错误，我在通过spark-submit运行spark作业时遇到问题：
16/11/16 11:41:12 ERROR yarn.ApplicationMaster: User class threw exception: java.lang.NoSuchMethodException: org.apache.hadoop.hive.ql.metadata.Hive.loadDynamicPartitions(org.apache.hadoop.fs.Path, java.lang.String, java.util.Map, boolean, int, boolean, boolean
浏览 6提问于2016-11-17得票数 0
我有一个用Java编写的Spark流媒体应用程序，使用Spark 2.1。我正在使用KafkaUtils.createDirectStream来阅读卡夫卡的消息。我正在使用kryo编码器/解码器来处理kafka消息。我在Kafka属性中指定了这一点-> key.deserializer、value.deserializer、key.serializer、value.deserializer
当Spark以小批量拉取消息时，消息使用kryo解码器成功解码。然而，我注意到Spark executor创建了一个新的kryo解码器实例，用于解码从kafka读取的每个消息。我通过将日志放在解码器构
浏览 0提问于2017-12-15得票数 6
我正在尝试在我的本地spark会话中启用动态分区(不是在应用程序模式下)
我在我的pyspark shell中运行以下命令(使用spark 2.4)
spark.sqlContext.setConf("hive.exec.dynamic.partition"，"true") spark.sqlContext.setConf("hive.exec.dynamic.partition.mode"，“非严格”)
低于错误AttributeError：'SparkSession‘对象没有属性'sqlContext’
浏览 16提问于2020-05-28得票数 0
回答已采纳
3回答
我之前已经成功地使用了pyspark for Spark Streaming (Spark 2.0.2)和Kafka (0.10.1.0)，但我的目的更适合于结构化流媒体。我尝试在线使用示例：
使用下面的类似代码：
ds1 = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "topic1")
.load()
浏览 2提问于2017-07-07得票数 1
2回答
我正在使用jceks文件解密我的密码，并且无法在纱线群集模式下读取加密密码。
我尝试过不同的方法，比如
spark-submit --deploy-mode cluster
--file /localpath/credentials.jceks#credentials.jceks
--conf spark.hadoop.hadoop.security.credential.provider.path=jceks://file////localpath/credentials.jceks test.py
spark1 = SparkSession.builder.appName("xy
浏览 4提问于2019-08-26得票数 1
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM
腾讯云开发者
