address:string, attention:string, city:string, state:string, zip:string >,submittedAt string "
from pyspark.sql.functions
浏览 8提问于2021-06-09得票数 0
我在一个spark会话中配置了几个结构化的流。我需要知道在每个流中读取和写入了多少条记录。例如，我有这两个流： read-s3 -> transform -> write-s3 read-s3 -> transform -> write-db 我知道如何使用SparkListener()、.onTaskEnd()，但在这一点上，我没有查询名，taskEnd.taskMetrics().outputMetrics().recordsWritten(
浏览 25提问于2018-02-01得票数 2
回答已采纳
我有一个两台计算机的EMR集群，安装了从s3读取数据的PySpark。代码是一个非常简单的过滤和转换操作，使用sqlContext.readStream.text从存储桶中获取数据。该存储桶大小约为10TB，由bucket/year/month/day/hour/*组织的大约75k个文件，其中*最多代表20个大小为128MB的文件。我通过提供存储桶s3://bucket_name/dir/并让PySpark读取其中的所有文件来启动<e
浏览 0提问于2018-02-20得票数 0
2回答
火花主
我按照以下文件中的具体步骤启动了kafka
from pyspark import SparkContext
from pys
浏览 1提问于2020-09-09得票数 0
回答已采纳
我正在尝试使用以下代码从流路径连接流Json文件 .option("maxFilesPerTrigger", 1).json(stream_path)) .partiti
浏览 1提问于2021-07-21得票数 2
我有一个星火结构流消费记录，从卡夫卡主题与2分区。 .queryName("query2").start()Problem：每次在两个分区中推送记录时例如，卡夫卡分区0-记录被推送，火花将处理query1。Kafka Partition1-当query1忙于处理时，记录被推送，spa
浏览 0提问于2020-07-24得票数 1
如何在DataFrame中为流PySpark设置架构。from pyspark.sql import SparkSessionfrom pyspark.sql.functionsimport splitfrom pyspark.sql.types import *
.bui
浏览 3提问于2016-12-29得票数 3
回答已采纳
bytesWritten = " + bytesWritten) }代码进入块，但是由recordsWritten编写的输入记录的值始终为dfb4fa939228", "timestamp" : "2018-07-30T14:20:33.486Z",
"batchId"
浏览 0提问于2018-07-25得票数 3
我只是使用querycassandra处理器查询cassandra表，但我不理解的是如何将Json输出文件作为输入文件传递给ExecutePyspark处理器，然后我需要将我的Spark数据传递给Hive。请帮我一下，谢谢。
火花特性：
浏览 1提问于2018-03-14得票数 3
回答已采纳
我正在做星火结构的流/kafka，这些都被记录在控制台中，但我不知道这一切意味着什么。45bb-ac20-68318e704e30", "timestamp" : "2019-08-04T22:02:08.870Z", "numInputRows} "endOffset" : {
"service-calls
浏览 0提问于2019-08-04得票数 2
回答已采纳
我有一个简单的Spark结构化流作业，它使用Kafka0.10API从Kafka读取数据并写入我们的S3存储。从日志中我可以看到，对于触发的每一批，流应用程序正在取得进展，并且正在消耗来自源的数据，因为endOffset大于startOffset，而且每一批都在增加。但是numInputRows始终为零，并且没有写入S3的行。 "name" : null,
&
浏览 0提问于2019-09-10得票数 0
回答已采纳
我不愿意在GAS中使用getConnection连接到Azure SQL数据库，因为我不确定连接是否会加密。因为这是我想要从GAS中提取的敏感数据，我想让它通过加密连接，但我不知道如何实现它？如果我将数据库托管在非Azure SQL Server上，我是否会有ore安全选项？getConnection提供了_serverSslCertificate、_clientSslCertificate和_clientSslKey，但我不认为它们可以在Azure SQL Server上使用
浏览 4提问于2018-01-07得票数 0
我正在尝试做的，是从我的事件中心读取一些数据，并将其保存在azure数据湖中。然而，问题是，流不会停止，writeStream步骤也不会触发。我找不到任何设置来确定输入速率何时达到0，以便在那时停止流。 ?
浏览 26提问于2021-07-10得票数 2
2回答
在使用建议的方法来回答问题时，代码无法添加多个事件处理程序。问题与将方法添加到TDelegateImpl<T>.Add()中的事件列表有关，TList<T>.IndexOf方法使用Compare方法查找现有方法，结果始终为0-这意味着TMethod的左和右是相同的Equals方法使用TMethod类型转换，并显式比较TMethod.Code和TMethod.Data，其中Compare转换为始终相同的地址。为什么在TList<T>.I
浏览 0提问于2011-09-15得票数 5
回答已采纳
2回答
在我的iOS应用程序中，我遇到了一些视觉限制。+-------------------++-------------------++-------------------+
一次只有一个View2、View3或View4是可见的，而其他的在不使用时是隐藏的。HeaderView为100，View2、View3和V
浏览 0提问于2016-06-03得票数 0
回答已采纳
我正在创建一个SSIS包来查询数据库并将结果提取到csv文件中。我正在使用2019。我将“标题行设置为跳过：”设置为0- 4。我对Preview选项卡中的数据行做了相同的操作，没有任何更改。数据流任务平面文件目标2:平面文件目标的输入列数。InputsFlat文件目标输入不能为零。
浏览 3提问于2022-07-06得票数 0
回答已采纳
我们有一个Azure SQL数据库，在那里我们收集了大量的传感器数据，我们定期从其中提取数据，并使用python脚本对其进行一点转换。最终结果是一个pandas DataFrame文件。我们希望将转换后的数据存储在Azure数据库中，并将其用作power BI仪表板的来源。有没有可能将熊猫的DataFrame转换成SQL，并将其存储在Data Lake上，然后从那里流式传输数据？我读到可以在data Lake上存储结构化数据，甚至可以对其进行查询，但我不确定这是否是最好的解决方案。 (我
浏览 15提问于2020-07-24得票数 0
回答已采纳
1回答
我在结构化流(spark 2.2.0)中使用了一个自定义接收器，并注意到spark为输入行数生成了错误的度量--它总是零。8710-78223a88edc7", "timestamp" : "2018-01-25T18:39:52.949Z", "logOffs
浏览 8提问于2018-01-27得票数 2
回答已采纳
要求:我需要在Spark (执行转换)中处理来自Kafka ( JSON格式)的流数据，然后将其存储在数据库中。("string"), schema))程序运行，但我在控制台上获得的值是9fc462e0-385a-4b05-97ed-8093dc6ef37b",
"name&qu
浏览 1提问于2017-04-08得票数 1
我正在尝试使用数据库中的pyspark从Heroku postgres数据库流，使用kafka作为插件。下面有这样的代码，但是它连续运行，没有返回任何内容，也没有错误。我尝试在代码中添加触发器，但不知道要使用什么导入，因为我没有在网上找到它的包。.{“消息”：“初始化源”，“isDataAvailable”：False，“isTriggerActive”：False}
我使用了kafka-python，并且能够读取流，但是使用pyspark<
浏览 14提问于2022-10-16得票数 0
云服务器
ICP备案
云直播
对象存储
实时音视频
腾讯云开发者
