2回答
我用readStream和进程文件读取目录中的文件，最后我有一个数据帧，我想把它写到oracle表中。api.here我的代码： def SaveToOracle(df,epoch_id): df.write.format('jdbc').options( driver='oracle.jdbc.driver.Oracle
浏览 147提问于2021-08-11得票数 0
我假设spark.sql(查询)是在使用spark.sql时使用的，而spark.read.format("jdbc").option("query“、"")则是在使用oracle语法时使用的。
浏览 3提问于2021-08-16得票数 3
回答已采纳
1回答
我想要记录从spark结构化数据流的传入流中读取到数据库的记录数量。我正在使用foreachbatch来转换传入的流批处理，并将其写入所需的位置。如果在特定的小时内没有记录，我想记录读取的0条记录。但是，当没有流时，foreach批处理不会执行。有人能帮我吗？我的代码如下： val incomingStream = spark.readStream.format("eventhubs").options(customEventhubParameter
浏览 10提问于2020-06-22得票数 1
从一个简单的Python脚本中，我每5秒将一些虚拟数据流到Kafka主题中。然后，使用Spark结构化流，我将这个数据流(每次一行)读入PySpark DataFrame中，其中包含startingOffset = latest。最后，我试图将这一行附加到已经存在的Cassandra表中。有一行数据正在成功地写入Cassandra表中，但我的问题是，每次覆盖时，都会被覆盖，而不是将追加到表的末尾。.opti
浏览 1提问于2020-10-21得票数 0
回答已采纳
我有一个火花结构化蒸汽作业，从kafka读取数据流，并写入到HDFS的输出。我的问题是，我需要一整天的汇总结果，直到特定的时间。既然spark structured不支持complete/update模式，有没有办法做到这一点？
如果我得到的数据是上午10点，我需要当前日期上午10点之前的汇总结果...
浏览 0提问于2019-01-19得票数 0
2回答
JDBC源的火花结构化流
、、、、
有人能让我知道是否有可能从JDBC源中触发结构化流吗？例如，SQL或任何RDBMS。
但是，我想知道它是否在Apache上得到了官方的支持？
浏览 8提问于2022-02-26得票数 1
回答已采纳
我正在寻找一个包，或者使用redshift作为结构化数据流的源的以前的实现。spark.readStream \ .option('url', redshift_url) \
.option('forward_spark_s3_credentials', 'true&#x
浏览 2提问于2020-11-18得票数 0
我有一个管道设置，从Kafka读取数据，使用Spark结构化流处理数据，然后将拼花文件写入HDFS。数据查询的下游客户端使用Presto配置，以便将数据作为Hive表读取。Kafka --> Spark --> Parquet on HDFS --> Presto
一般来说，这是可行的。当Spark作业运行批处理时发生查询时，就会出现问题。我希望能够连续写入适当的HDFS文件夹，而
浏览 3提问于2017-11-16得票数 1
回答已采纳
1回答
火花结构化流检查点在生产中的使用
、、、、
当使用Spark结构化流时，我难以理解检查点是如何工作的。
将已连接的DataFrame写入HBase表。在每次运行此代码时，我都希望只<em
浏览 7提问于2020-07-08得票数 3
我正在使用结构化流来评估一个15节点星火集群中的特定数据流。我在我的应用程序中定义了两个流查询：
SQ2 -从HDFS (与上面相同的路径)读取数据作为文件流，->使用ForeachWriter进一步处理->写入外部数据库。我的问题--我是否在这里看到了一个竞争条件，在这里，SQ2从HDFS获取部分
浏览 3提问于2018-07-13得票数 4
1回答
我有一个连接到Apache Kafka数据流作为源的Spark 2.0.2结构化流作业。该工作从Kafka获取Twitter数据(JSON)，并使用CoreNLP为数据添加情感、词性标注等。然而，当我设置一个独立的Spark集群时，只有一个worker被用来处理数据。我有两个具有相同能力的工人。提前感谢您给我们指明了正确的方
浏览 0提问于2016-12-06得票数 1
我设置了一个从Kafka主题读取的Spark结构化流查询。如果在Spark查询运行时更改了主题中的分区数量，Spark似乎不会注意到，并且不会消耗新分区上的数据。除了停止和重新启动查询之外，有没有办法让Spark检查同一主题中的新分区？spark .format("kafka")
.o
浏览 32提问于2019-11-08得票数 0
1回答
我开始学习火花，并有一个困难的时间理解背后的合理性结构化流在星火。结构化流将到达的所有数据视为无界输入表，其中数据流中的每个新项都被视为表中的新行。val spark = SparkSession.builder.appName("SimpleApp").getOrCreate()
val csvSchema = new StructType()string").add("price", "string").add(
浏览 2提问于2017-05-20得票数 5
回答已采纳
我使用JDBC connection使用Spark连接到oracle database，并尝试读取包含4,000,000行的oracle表。在启动spark-shell/submit时，我为每个执行器使用了30个执行器、5个执行器内核和4g内存。当读取数据帧的计数或尝试写入数据时，它只使用一个执行器从oracle读取/写入数据。命令片段：-
spark-shel
浏览 0提问于2020-10-30得票数 0
我使用结构化数据流从Kafka流式传输数据 .readStream .option("kafka.bootstrap.servers"group.id", UUID.randomUUID().toString) .load()
然后尝试使用Cassandra表</
浏览 0提问于2018-11-16得票数 1
1回答
TimestampType(), True), StructField("action", StringType(), True) ])
df = spark.readStream.format在最后一步，我碰到了一个错误：
我们如何解决这个问题？
浏览 1提问于2020-05-13得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
