Spark Dataframe写入google pubsub

文章/答案/技术大牛

发布

1回答

apache-spark、apache-spark-sql、google-cloud-pubsub、google-cloud-dataproc

我正在尝试通过Dataproc集群上的Spark将Parquet文件写入Pubsub。我使用了下面的伪代码 dataFrame .foreachPartition(partition => { val topicName我没有找到任何有关将数据帧写入pub-sub的相关信息。有什么建议吗？usercache/root/appcache/application_1608332157194_002

浏览 208提问于2020-12-29得票数 1

1回答

从google pubsub到spark streaming的数据摄取速度很慢

apache-spark、google-bigquery、spark-streaming、google-cloud-pubsub、google-cloud-dataproc

我正在使用google cloud Dataproc Spark集群来运行Spark streaming作业，该作业从多个PubSub订阅中读取数据并写入BigQuery。// "org.apache.spark" %% "spark-hive" % sparkVersion, "com.google.cloud" % "google

浏览 5提问于2019-07-22得票数 3

1回答

Apache Spark 2.2.1中的谷歌PubSub

apache-spark、google-cloud-pubsub

我正在尝试在Spark应用程序中使用谷歌云PubSub。为了简单起见，我们只说这个应用程序是Spark的shell。然而，通过这样一个简单的设置(只有Spark和一个Google Cloud PubSub依赖项)，我想不出如何解决这个问题。bash-4.4# spark-shell --packages com.google.cloud:google-cloud-pubsub:1.105.0 Ivy Default

浏览 17提问于2020-05-07得票数 0

2回答

如何将数据从Google PubSub主题流到PySpark ( Google上)

python、pyspark、streaming、google-cloud-pubsub

我在Google PubSub中有一个主题的数据流。message.ack() future.result() 上面的python代码从GooglePubSub主题(使用订阅服务器subscriber_name)接收数据，并按照预期将其写入终端。我希望将来自主题的相同数据流到PySpark (RDD或dataframe)中，这样我就可以进行其他流转换，如PySpark中

浏览 1提问于2018-09-17得票数 1

1回答

使用多个jars创建dataproc集群

google-cloud-platform、google-cloud-pubsub、google-cloud-dataproc

我正在尝试创建一个dataproc集群，它将dataproc连接到pubsub。:spark.jars=gs://bucket/jars/spark-streaming-pubsub_2.11-2.4.0.jar,gs://bucket/jars/google-oauth-client-1.31.0.jar,gs://bucket/jars/google-cloud-d

浏览 8提问于2021-11-27得票数 2

回答已采纳

1回答

如何使用Google电子表格执行Spark Streaming？

python、google-sheets、spark-streaming、read-write、real-time-data

需要实时提供的输入，并且是google电子表格的形式(多个用户同时提供数据)。此外，需要将代码的实时输出写回其相邻列中的电子表格。也请帮帮我。谢谢

浏览 17提问于2020-11-05得票数 0

1回答

Google Dataproc with Jupyter -下载notebook生成的文件

jupyter-notebook、google-cloud-dataproc

我们正在使用Google Cloud Dataproc进行快速数据分析，并且我们经常使用Jupyter笔记本。对于我们来说，一种常见的情况是生成一份报告，然后我们希望将其下载为csv。FileLink实现，例如： from IPython.display import FileLinksFileLinks(path) 这不适用于Dataproc，因为笔记本保存在Google

浏览 19提问于2019-01-13得票数 0

2回答

如何使用python客户端库获取Pubsub中存在的未送达消息(指标api)的数量？

python、google-cloud-platform、stackdriver、google-cloud-stackdriver

我正在使用Pubsub作为一个排队机制工具，我想知道驻留在Pubsub主题中的消息的计数。出于同样的目的，我决定使用Google API指标pubsub.googleapis.com/subscription/num_undelivered_messages，但我不知道如何使用python客户端库from google.cloud import monitoring_v3os.environ["GOOGLE_APPLICATI

浏览 6提问于2019-04-10得票数 2

回答已采纳

1回答

Spark :将Array[Byte]数据转换为RDD或DataFrame

scala、apache-spark

我有ArrayByte格式的数据，我想将其转换为Spark RDD或DataFrame，这样我就可以将数据以文件的形式直接写入Google存储桶中。我无法将ArrayByte数据直接写入Google bucket。所以寻找这个转换。我的以下代码能够将数据写入本地文件系统，但不能写入Google存储桶 val encrypted = encrypt(original, readPublicKey(pubKey), outFile,Str

浏览 48提问于2019-11-02得票数 0

回答已采纳

1回答

用于创建com.google.common.base.Preconditions.checkArgument Publisher异常的Scala Spark代码: java.lang.NoSuchMethodError

scala、apache-spark、google-cloud-platform、sbt、google-cloud-pubsub

" %% "spark-core" % "2.3.2" % "provided", "com.google.cloud" % "google-api-services-bigquery" % "v2-rev456-1.25.0&q

浏览 80提问于2020-02-16得票数 2

回答已采纳

1回答

Spark Streaming:在Google Pubsub上注册自定义接收器kryo

apache-spark、google-cloud-pubsub、kryo

我使用的是带有Kryo序列化的Spark 2.0.2。我正在尝试实现一个自定义接收器，用于将来自谷歌PubSub的消息摄取到Spark Streaming中：group (java.util.concurrent.Executors$DefaultThreadFactory) val$backin

浏览 0提问于2017-04-25得票数 1

1回答

是否可以使用数据流将重复的pubsub消息删除回pubsub？

message-queue、google-cloud-dataflow、apache-beam、google-cloud-pubsub

我有一个将数据写入Google Cloud pubsub的应用程序，根据pubsub的文档，由于重试机制而导致的重复是偶尔会发生的事情。还有失序消息的问题，这在pubsub中也没有保证。此外，根据文档，可以使用Google Cloud Dataflow对这些消息进行重复数据删除。我想让这些消息在消息队列(即云pubsub)中可用，以供服务使用，并且云数据流似乎确实有一个pubsubio编写器，然而，您不会回到写入pubsub可能会创建

浏览 46提问于2019-03-12得票数 2

1回答

如何从"Google函数“连接到"Google”公共数据集

python-3.x、google-cloud-platform、google-bigquery、google-cloud-functions、google-cloud-iot

我修改了"Google“服务的默认hello_pubsub (Python3.7)函数，以连接到我在"Google”服务中创建的dataset表。有人，请帮帮我import sqlite3 """Triggered fromcontext (google.cloud.functions.Context): Metadata for th

浏览 1提问于2019-07-31得票数 2

回答已采纳

1回答

Dataproc:使用BigQuery读写数据时使用PySpark时的错误

python、pyspark、google-bigquery、google-cloud-dataproc

') \# print("DataFrame shape: ", (df.count(), len(df.columns))) # ------- CONTINGEN

浏览 15提问于2022-08-09得票数 1

回答已采纳

1回答

Couchbase Spark Connector支持PySpark吗？

dataframe、pyspark、spark-dataframe、couchbase、pyspark-sql

我们有10个节点的AWS EMR集群，emr 5.5.0版本，Spark 2.1.0我们希望将此摘要数据(PySpark DataFrame)写入couchbase数据库。Couchbase Spark Connector支持PySpark吗？如果是，请分享一下如何使用PySpark将数据写入</

浏览 16提问于2017-08-05得票数 0

1回答

将spark数据帧写入固定宽度文件java spark

java、apache-spark、apache-spark-sql

我已经使用java spark dataframe将CSV读取到dataframe中，现在我必须对每个列应用一些宽度，并将数据写入固定宽度的文件中。谁能解释一下如何使用java spark将dataframe中的数据写入到固定宽度的文件中？我只需要java spark中的解决方案

浏览 0提问于2020-11-19得票数 0

3回答

在数据文件上使用partitionBy时对dataproc的警告

apache-spark、pyspark、google-cloud-platform、google-cloud-dataproc

我正在尝试使用dataproc编写一个数据内容到google云存储中。虽然写入是成功的，但在我下面粘贴的日志中有很多警告消息。在创建集群时，还是在pyspark程序中，是否缺少一些我需要的设置？备注：由dataframe在google存储上写入的数据是> 120 GB未压缩的。但我注意到了同样的警告，即使我处理的数据大小为1GB，未压缩。这是一个简单的dataframe，有50列被读取，一些转换被完成并写入磁盘。Dataframe写语句如下所示：

浏览 0提问于2018-04-02得票数 2

回答已采纳

1回答

-从PubSub到Parquet

python、google-cloud-platform、google-cloud-dataflow、apache-beam

我试图使用将谷歌的PubSub消息写入。PubSub消息以json格式出现，我要执行的唯一操作是从json到parquet文件的转换。在官方文档中，我找到了一个由google提供的模板，它从Pub/Sub主题中读取数据，并将Avro文件写入指定的云存储桶()。问题是模板源代码是用Java编写的，而我更喜欢使用Python。

浏览 0提问于2020-07-21得票数 0

回答已采纳

2回答

使用已知模式保存空DataFrame* (Spark 2.2.1)*

apache-spark、parquet、databricks

是否可以使用已知的模式保存一个空的DataFrame，以便将该模式写入该文件，即使该文件没有任何记录？def example(spark: SparkSession, path: String, schema: StructType) = { val dataframeWriter = dataframe</e

浏览 2提问于2018-04-14得票数 5

1回答

如何将流xml加载到BigQuery中的指南

xml、google-bigquery

如果有人能在这里提供帮助，我将不胜感激，我们刚刚开始研究GCP，需要一个健壮且简单的模式来将发布在云发布/订阅上的xml格式的事务数据加载到一个日期分区的BigQuery表中，以便在AirFlow编排的复杂下游批处理中使用。以前有人这么做过吗？为了在摄取端允许模式漂移，一种选择是将xml转换为json，并将json存储为字符串，并在顶部使用json函数提取字段以进行下游处理，这种方法的优缺点是什么？在我们的例子中，一个优点是在xml中有很多(300+)字段，但最初只使用了一个子集，但随着时间的推移，我们需要能够快速地“打开”新字段。也许可以更进一步，将原始xml存储在BQ中，并使用B

浏览 22提问于2020-06-29得票数 0

点击加载更多