如何使用 Scala 将表更快地加载到 Spark 数据帧中？
我编写了一个代码，该代码应该使用 Spark 将许多表（通过 LTables 方法列出）加载到 Scala 中的不同数据帧中。 这是我的代码： LTables.iterator.fore…
在 Spark 结构化流中解释数组 JSON 中的数据帧
我的数据框中有一列下面的 json 字符串，如何分解/压平它以获得单级数据框？ 目前的模式是 df |-json_data (StringType) 如何拥有以下模式的 df ？ df…
在 Spark 结构化流中计算滑动窗口中的多个聚合
我有一个流源，它发送事件，其中每条记录由 3 个字段组成（CreationTime、FP、Detected） 这里，“FP”代表误报。 “FP”和“检测到”字段的值可以为 …
Spark中的saveAasTable和save有什么区别
我正在使用 Pyspark，并且想要将分区插入并覆盖到现有的配置单元表中。 在这个用例中 saveAsTable() 不合适，它会覆盖整个现有表 insertInto() 的行为…
“记录并跳过”的正确方法是什么？ Spark-Streaming 中经过验证的转换
我有一个火花流应用程序，我想在主要操作之前进行一些数据转换，但转换涉及一些数据验证。 当验证失败时，我想记录失败案例，然后继续处理其余的事情…
Spark.readStream 与 Kafkautils.createDirectStream
我想知道是否有人知道这两种语法之间的区别是什么？我知道两者都用于从 Kafka 读取数据，但它们有什么区别？ Spark.readStream.format("kafka") Kafka…
警告 CSVHeaderChecker：CSV 标头不符合架构。 - 但标题是正确的
我正在尝试使用 Spark 流式传输 CSV 文件。 我受到 https://dzone.com/articles/spark-structed 的启发-streaming-using-java。 但是我收到错误： 22/…
覆盖模式下的 pyspark insertInto 是追加而不是覆盖分区
我是一名正在开发 Spark 2.3 的数据工程师，我遇到了一些问题： 将 Spark.conf 更改为“动态” spark = spark_utils.getSparkInstance() spark.conf.s…
Cosmos Changefeed Spark 流随机停止
我有一个 Spark 流作业，它读取 Cosmos Changefeed 数据，如下所示，在具有 DBR 8.2 的 Databricks 集群中运行。 cosmos_config = { "spark.cosmos.ac…
sparkstreaming local[4]消费Kafka四分区topic，只有一个消费者在工作
sparkstreaming消费Kafka，用local[4]模式处理有四个分区的topic，为什么只起来一个消费者，手动维护的四个分区offset更新值都一样，有碰到过这种情况…
Spark on yarn 在创建Hbase的Connection时，报 ClassNotFoundException
ConnectionFactory.createConnection(configuration) 在执行以上方法时候报的错误:Caused by: java.lang.ClassNotFoundException: org.apache.hadoop…
python开发，spark接收kafka的输入流时
import cv2 import sys import findspark findspark.init() from kafka import KafkaConsumer from pyspark import SparkContext from pyspark.strea…
