(DataSource.scala:115) at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:232)
at org.apache.spark.sql.streaming</
浏览 18提问于2021-10-17得票数 4
回答已采纳
1回答
如何节流火花流？
、、
这个问题脱离了我关于管理AmazonDynamoDbClient节流和重试的另一个问题。然而，我认为解决方案可能存在之前，我甚至到发电机电话。我的高级过程如下:我有一个scala应用程序，它使用Apache读取大型CSV文件，并对它们执行一些聚合，然后将它们写入dynamo。我将其部署到EMR以提供可伸缩性。问题是，一旦聚合完成，我们就有数以百万计的记录准备进入发电机，但我们有一个写容量的发电机。它们不需要立即插入，但是最好能控制每秒多少次，这样我们就可以根据用例对其进行微调。下面是我到目前为止所得到<e
浏览 3提问于2017-03-28得票数 3
1回答
我有一个S3 json数据集，它是KMS客户端加密DynamoDB的转储(即每条记录都是独立加密的KMS客户端)。我想知道是否有什么办法限制这些星图操作？
浏览 7提问于2017-05-13得票数 1
1回答
在过去的几个月里，我使用了相当多的结构化流媒体来实现Stream Jobs (在使用了很多Kafka之后)。在阅读了“使用Apache Spark进行流处理”一书后，我有了这样一个问题:有没有什么要点或用例可以让我使用Spark流而不是结构化流？我是不是应该花点时间去了解它，或者既然我已经在使用Spark Structured Streaming了，我应该坚持使用它，并且之前的API没有任何好处。将非常感谢任何意见&
浏览 17提问于2020-04-06得票数 2
是否可以限制Kafka消费者返回Spark Streaming的批量大小？
我之所以问这个问题，是因为我得到的第一批记录有上亿条记录，需要很长时间才能处理和设置检查点。
浏览 0提问于2016-10-11得票数 18
回答已采纳
我想直接在代码中覆盖spark.sql.shuffle.partitions参数： val sparkSession = SparkSession .appName("SPARK")
sparkSession.conf.set("spark.sql.shuffle.partitions", 2) 但此设置不会生效，因为在日志中我收到以下警告消息虽然在spark-submit外壳<em
浏览 39提问于2021-04-30得票数 2
我正在阅读Spark2.2.0文档，发现这是对Static DataFrames 的奇怪引用
因为Spark2.0，DataFrames和数据集可以表示静态的、有界的数据，以及流的、无界的数据。与静态数据集/DataFrames类似，您可以使用公共入口点SparkSession (Scala/Java/Python/R )从流源创建流DataFrames/Datas
浏览 1提问于2017-12-24得票数 1
回答已采纳
为了实现这一点，我提供了一个硬编码的JSON模式作为StructType。我正在寻找一种在流过程中动态推断主题模式的好方法。这是我的代码：(这是Kotlin，而不是通常使用的Scala) .readStream() .option("kafka.bootstrap.serversProcessingTime("25 seconds"))
.start("
浏览 2提问于2020-10-21得票数 1
回答已采纳
1回答
(read_kafka_data.py)中读取来自以下主题的数据： schema = StructType([StructField("col1", StringType("spark.streaming.backpressure.enabled",”true”)
启用或禁用火花流的内部背压机制(自1.5起)。这使得火花流能够根
浏览 0提问于2021-09-13得票数 4
回答已采纳
我有一个星火结构的流： .readStream .option("kafka.bootstrap.serverswriteStream .format("parquet")但是在data文件夹中
浏览 5提问于2017-05-25得票数 4
回答已采纳
我们有一个在Spark2.3.3上运行的星火流应用程序 kafka_stream = spark \ .format("kafka我们认为可能有一种方法可以用一些参数限制第一批的大小，但是我们没有发现任何有帮助的东西。我们试过：
spark.streaming.kafka.maxRatePerPartition=1000，s
浏览 1提问于2019-04-02得票数 10
回答已采纳
我的理解是Spark structured Streaming是建立在Spark SQL之上的，而不是Spark streaming。因此，下面的问题是，应用于spark流的属性是否也适用于spark结构化流，例如：
spark.streaming.backpressure.initialRate spark.strea
浏览 1提问于2018-09-03得票数 1
火花流中有spark.streaming.kafka.maxRatePerPartition属性，它限制每秒从Apache读取消息的数量。斯托姆也有类似的财产吗？
浏览 2提问于2016-06-10得票数 0
回答已采纳
我想使用星火流，并连接它与卡夫卡。然而，我仍然得到了NoSuchMethodError: NoSuchMethodError错误，现在我只是不知道下一步该做什么。我的设置：Scala 2.11火花2.2.1我甚至不能运行示例脚本$spark$sql$execution$streaming$StreamExecution$$runBatches(StreamExecution.scala:2
浏览 1提问于2018-02-09得票数 3
1回答
我有一个Spark/Scala应用程序，这里我的需求是在一个目录中查找一个文件并对其进行处理，最后清理该目录。难道不可以在spark应用程序中这样做吗- When it finds the file continue the process我们目前使用一个外部应用程序来执行此file-watching process操作，因此为了删除该third-party application上的
浏览 0提问于2019-03-21得票数 0
我使用在亚马逊网络服务中运行的spark独立集群(spark和spark-streaming-kafka版本1.6.1)，并使用检查点目录StreamingContext.getOrCreate(config.sparkConfig.checkpointDir, createStreamingContex
浏览 2提问于2016-12-19得票数 2
1回答
我想在用Scala编写的Nifi自定义处理器中创建Spark会话，到目前为止，我可以在scala项目上创建spark会话，但是当我在nifi自定义处理器的OnTrigger方法中添加这个spark会话时，spark会话永远不会创建，有什么方法可以实现这一点吗？到目前为止，我已经导入了spark-core和spark-sql库，欢迎任何反馈。
浏览 4提问于2021-03-10得票数 0
我有一条从卡夫卡读到GCP的管道。档案记录太少了。我想创建更大的文件。到目前为止，这是我如何配置梁(至少我认为是一个相关的参数)。我的问题是如何控制从束流管道产生的文件的大小？numShards: 0 checkpointDurationMillis: 30000以下是与流相关的火花配置参数=120
浏览 3提问于2021-05-26得票数 0
回答已采纳
我试图读取卡夫卡的数据，并将其上传到格林梅利的数据库使用火花。我使用的是格林梅-火花连接器，但我正在获取数据源，io.pivotal.greenplum.spark.GreenplumRelationProvider不支持流写入。是否格林梅源不支持流媒体数据？我可以在网站上看到“连续ETL管道(流)”。我曾尝试将数据源命名为“绿梅”，并将"io.pivotal.greenplum.spark.GreenplumRelationProvider“转换为.form
浏览 0提问于2019-04-04得票数 0
回答已采纳
云服务器
ICP备案
实时音视频
即时通信 IM
云直播
腾讯云开发者
