高效地将Spark中的数据帧发送到其他应用程序
可以通过使用Spark的数据源API和适当的数据格式来实现。
数据源API是Spark提供的一种机制，用于将数据从不同的数据源加载到Spark中，并将数据从Spark中导出到其他应用程序。对于将数据帧发送到其他应用程序，可以使用数据源API中的输出数据源接口。
以下是一种可能的实现方式：
- 首先，需要选择适当的数据格式来存储和传输数据帧。常见的数据格式包括Parquet、Avro、JSON、CSV等。选择数据格式时需要考虑数据的结构、大小、读写性能等因素。
- 使用Spark的数据源API将数据帧保存到选定的数据格式中。可以使用
write方法将数据帧写入指定的数据格式。例如，使用Parquet格式保存数据帧的示例代码如下：
dataFrame.write.format("parquet").save("path/to/save")- 在其他应用程序中，可以使用适当的库或工具来读取选定的数据格式。例如，使用Python中的Pandas库可以读取Parquet格式的数据帧：
import pandas as pd
data_frame = pd.read_parquet("path/to/save")- 如果需要将数据帧发送到远程应用程序，可以使用网络通信协议（如HTTP、TCP/IP等）将数据传输到目标应用程序。具体的实现方式取决于应用程序之间的通信方式和协议。
在腾讯云中，可以使用腾讯云对象存储（COS）作为数据源来保存数据帧。腾讯云COS是一种高可用、高可靠、低成本的云存储服务，适用于存储和处理大规模数据。您可以使用腾讯云COS的SDK或API来将数据帧保存到COS中，并在其他应用程序中读取数据。
腾讯云COS产品介绍链接地址：https://cloud.tencent.com/product/cos
请注意，以上只是一种实现方式的示例，具体的实现取决于您的需求和环境。在实际应用中，还需要考虑数据的安全性、性能优化、错误处理等方面的问题。
相关·内容
我想将Spark上下文中的数据帧发送到我的客户端应用程序。目前，我只是将数据帧转换为内存中的列表，并通过ZMQ将该列表发送给客户端。但是调用Dataframe.collectAsList()需要花费很多时间。有什么方法可以提高这个过程的速度吗？或者我应该尝试其他方式将数据从Spark传递到其他应用程序？提前谢谢。
浏览 7提问于2016-09-13得票数 2
回答已采纳
假设我有一个名为df的熊猫数据帧
id value1 value2
1 2 1
2 2 1
3 4 5
在普通的Python中，我编写了一个函数来处理此数据帧并返回字典：
d = dict()
for row in df.itertuples()
x = do_something (row)
d[x[0]] = x[1:]
我正在尝试使用Spark重新实现这个函数。
d = dict() # define a global var
def do_something (id, value1, value2):
# business logic
d[x0] = [x1,x2
浏览 4提问于2017-08-25得票数 0
我在Spark里有两个数据帧。它们都有一个唯一的id。结构如下所示
df1:
id_df1 values
abc abc_map_value
cde cde_map_value
fgh fgh_map_value
df2:
id_df2 array_id_df1
123 [abc, fgh]
456 [cde]
我想要获取以下数据帧结果：
result_df:
id_df2 array_values
123 [map(abc,abc_map_value), map(fgh,fgh_map_value)]
456
浏览 1提问于2017-05-02得票数 0
1回答
我使用星火在本地模式和一个简单的连接是太长。我获取了两个数据文件:a (8列和230万行)和B(8列和120万行)，并使用A.join(B,condition,'left')连接它们，最后调用了一个操作。它创建一个包含三个阶段的单个作业，每个阶段用于两个数据文件提取，另一个用于加入。令人惊讶的是，提取数据的阶段A大约需要8分钟，而数据B则需要1分钟。几秒钟之内就会加入。我的重要配置设置是：
spark.master本地的*
spark.driver.cores 8
spark.executor.memory 30g
spark.driver.memory 30g
浏览 0提问于2016-12-25得票数 5
回答已采纳
我使用Spark已经有一段时间了，在Python中取得了成功，但是我们有一个用C#编写的产品，它将从分布式和并行执行中受益良多。我做了一些研究，并尝试了新的C#应用程序接口，但目前这有点限制。
关于Ignite，从表面上看，它似乎是一个不错的选择。它得到了良好的.NET支持，它具有集群能力和跨网格分布计算的能力。
然而，我想知道在我们的用例中，它是否真的可以用来取代Spark --我们需要的是一种执行数据帧类型操作的分布式方式。特别是，我们在Python中的许多代码都是使用Pandas UDF实现的，我们让Spark负责数据传输和结果合并。
如果我想使用Ignite，我们的数据实际上更像是一个
浏览 14提问于2019-11-08得票数 0
回答已采纳
我正在尝试找出如何最好地存储和检索数据，从S3到Cassandra，使用Spark:我在Cassandra中存储了日志数据。我使用DSE运行Spark来执行数据分析，它工作得很好。日志数据每天都在增长，在任何给定的时间，我只需要在Cassandra中使用两周。我仍然需要在某个地方存储较旧的日志至少6个月，经过研究，带有Glaciar的S3看起来是最有希望的解决方案。我想使用Spark来运行一个日常作业，该作业查找第15天的日志，将它们从Cassandra中删除，然后将它们发送到S3。我的问题是:我似乎无法确定正确的格式来将Cassandra行保存到文件中，这样有一天我可能会将文件重新加载到Sp
浏览 0提问于2015-07-18得票数 4
我正在用Scala在Spark 2.3中做一些功能工程。
我在Spark DataFrame的一列中有IP地址，如下所示
然后，我使用data.groupBy("ip").count()获取每个IP地址的频率列表。这看起来像是
现在，我希望将这些频率中的每个频率映射到原始数据帧。在那里我会
ip | freq |
-- | |
123 | 3 |
567 | 7 |
857 | 10 |
123 | 3 |
解决此类问题的有效方法是什么？
浏览 0提问于2018-04-03得票数 0
如何将熊猫数据帧发送到hive表？
我知道如果我有一个spark数据帧，我可以将它注册到一个临时表中，使用
df.registerTempTable("table_name")
sqlContext.sql("create table table_name2 as select * from table_name")
但是当我尝试对registerTempTable使用pandas dataFrame时，我得到了以下错误：
AttributeError: 'DataFrame' object has no attribute 'regis
浏览 0提问于2016-04-28得票数 9
回答已采纳
在我的Spark Streaming作业中，CPU未得到充分利用(仅5-10%)。
它从Kafka获取数据并发送到DynomoDB或thridparty端点。
假设端点不是瓶颈，是否有任何建议的作业可以更好地利用cpu资源。
浏览 1提问于2017-03-18得票数 0
我有一个可以收听Kafka主题的Spark流媒体应用程序。当获得数据时，我需要对其进行处理，并将其发送到Kudu。目前，我正在使用org.apache.kudu.spark.kudu.KuduContext API，并对数据框调用插入操作。为了从我的数据创建数据框，我需要调用collect()，以便可以使用sqlContext创建数据框。
有没有一种方法可以在不调用collect()的情况下创建数据帧/将数据插入到Kudu中？
我们使用的是Spark 1.6
浏览 7提问于2018-08-08得票数 1
1回答
使用Spark流的红移
、、、
我有一个Kafka - Spark流应用程序，可以每分钟摄取和处理60K个事件。我需要一个数据库来存储我转换后的数据帧，以供可视化层访问。Redshift可以与Spark Streaming一起使用吗?还是应该使用Cassandra？我将在每30秒的spark窗口中处理和存储数据帧。另外，我需要从每个窗口的数据存储中读取数据。我猜Redhsift主要是一个数据仓库数据库，而不是用于OLTP类型的处理。有什么想法吗？
浏览 0提问于2017-09-28得票数 0
我有个疑问，我们什么时候广播数据帧。
广播的数据帧的副本被发送到每个执行器。
那么，Spark什么时候从每个Executor中驱逐这些副本呢？
浏览 4提问于2020-12-30得票数 0
我必须使用Spark从HDFS加载一个CSV文件到DataFrame中。我想知道是否有一个“性能”改善(查询速度)从一个DataFrame支持的CSV文件和一个支持一个拼花文件？
通常，我会像下面这样将CSV文件加载到数据帧中。
val df1 = sqlContext.read
.format("com.databricks.spark.csv")
.option("header", "true")
.option("inferSchema", "true")
.load("hdfs://box/
浏览 4提问于2016-09-16得票数 8
回答已采纳
我正在尝试选择出现在foo中但不出现在bar中的所有元素。我使用的是以下代码：
foo.repartition(1)
foo.cache()
bar.repartition(1)
bar.cache()
foo.select("col_1").except(bar.select("col_1"))
有没有更好或更快的方法来做这件事？当在集群上运行时，这比15+分钟花费的时间更多。
附加信息: foo将有大约100-1000个元素。bar将有40个million+元素。foo是一个数据帧，由使用Spark SQL从hive表(50列)读取的数据组成。bar是一个数
浏览 0提问于2019-05-02得票数 1
https://github.com/jupyter-incubator/sparkmagic/blob/master/examples/Send%20local%20data%20to%20Spark.ipynb 我已经看过上面关于如何从本地发送到spark内核的文档。 然而，我遇到了一个问题，我有60,000行的本地熊猫，但是当我在这个数据帧上尝试send_to_spark单元魔术时，只发送了2500行。有没有什么办法，我可以发送整个60,000行，而不拆分本地的df？ 我知道从spark发送到本地，%%spark -o df -n 60000就可以了。
浏览 23提问于2021-04-12得票数 0
回答已采纳
我有一个应用程序，它创建一些数据帧，将它们写入磁盘，然后使用vertica_python运行一个命令，将数据加载到Vertica中。由于加密驱动器，Spark Vertica连接器无法工作。
我想要做的是，让应用程序运行命令来加载数据，然后立即转移到下一个作业。然而，它正在做的是等待在Vertica中完成加载，然后再移动到下一个作业。我怎么才能让它做我想做的事呢？谢谢。
这个问题的奇怪之处在于，我希望在后台运行的命令就像db_client.cursor.execute(command)一样简单。这在正常情况下不应该被阻止，那么为什么它会出现在Spark中？
具体地说，发生的情况是我正在读取数据
浏览 0提问于2018-05-22得票数 0
我有一个只有数据但没有列名的CSV文件，现在我想用这个CSV文件的数据在Spark中创建一个数据帧，并为它创建模式(列名和数据类型)。我的代码如下： import org.apache.spark.sql.SQLContext
val sqlContext = new SQLContext(sc)
val employee = sqlContext.read.format("com.databricks.spark.csv")
.option("header", "false")
.option("inferSchema",
浏览 52提问于2019-02-26得票数 0
2回答
函数将R类型转换为星点类型。
、、、、
我有一个R数据框架，我想把它转换成远程集群上的星火数据帧。我已经决定将我的数据框架写到一个中间的csv文件中，然后使用sparklyr::spark_read_csv()读取该文件。我这样做是因为数据帧太大，不能直接使用sparklyr::sdf_copy_to()发送(我认为这是由于Livy中的限制)。
我想通过编写一个函数来编程地将数据帧中使用的R列类型传递到新的spark数据框架，该函数返回一个命名向量，我可以在spark_read_csv()中的spark_read_csv()参数中使用该向量。
浏览 0提问于2019-03-28得票数 0
回答已采纳
2回答
当我们创建spark数据帧时，我们将数据帧中的数据发送到Kudu和Kafka(依次被提取并进入S3)
然而，一旦数据帧被插入，我们在Kudu中看到的精度是1e-15
现在，如果我使用相同的数据帧并将其转换为Kafka可以使用的格式
override def getKafkaDataFrame(df: DataFrame) : DataFrame = {
df.withColumn("key", to_json(struct(df.col(PK1),
df.col(PK2)))
.withColumn("value", to_json(struct(
浏览 51提问于2020-02-07得票数 0
我正在准备我的数据(52 my )，以便在我的本地机器上执行一些范围查询。
我的数据在BSON文件中。我正在将它转换为spark rdd/dataFrame，并编写Cassandra用于快速范围查询。
我拥有的数据没有任何唯一的范围选择方法，所以我通过调用monotically_increase()并将其写入Cassandra，在rdd数据帧中添加了一个唯一的column(idx)。
但是Cassandra，正在将idx的值改写为非常大的值。
train_df = train_df.withColumn("idx", monotonically_increasing_id())
浏览 0提问于2017-11-19得票数 0
在关闭连接时，以以下方式复制到spark的数据帧会发生什么情况？
library(sparklyr)
library(dplyr)
sc <- spark_connect(master = "local")
iris_tbl <- copy_to(sc, iris)
spark_disconnect(sc)
如果它们不是自动删除的，除了通过下面的方法删除每个数据帧之外，有什么简单的方法可以删除在会话期间创建的所有数据帧吗？
sc %>% spark_session() %>% invoke("catalog") %>% invoke
浏览 5提问于2018-08-24得票数 0
回答已采纳
如何有效地将大型本地数据帧转换为SparkR数据帧？在我的本地开发机器上，当我尝试将~ 650MB的本地数据帧转换为SparkR数据帧时，它很快就会超出可用的内存，而我的开发机器上有40 of的Ram。
library(reshape2)
years <- sample(1:10, 100, replace = T)
storms <- sample(1:10, 100, replace = T)
wind_speeds <- matrix(ncol = 316387, nrow = 100,
data = sample(0:25
浏览 0提问于2016-09-08得票数 3
我有一个数据集，如下所示：
val
1
1
3
4
6
6
9
...
我无法将其加载到熊猫数据帧中，因为它太大了。因此，我使用Spark聚合数据以形成：
val occurrences
1 2
3 1
4 1
6 2
9 1
...
并将其加载到熊猫数据帧中。"val“列不超过100，所以它不会占用太多内存。
我的问题是，我不能很容易地在这样的结构上操作，例如使用pandas找到均值或中位数，也不能使用seaborn绘制箱线图。我只能使用我自己编
浏览 0提问于2018-09-18得票数 1
使用Spark 2.3，我知道我可以像这样读取JSON文档文件：
{'key': 'val1'}
{'key': 'val2'}
有了这个：
spark.json.read('filename')
当JSON文档之间没有换行符时，我如何将以下内容读入到数据帧中？
下面是一个示例输入。
{'key': 'val1'}{'key': 'val2'}
为了清楚起见，我希望得到一个两行的数据帧(frame.count() == 2)。
浏览 0提问于2018-07-13得票数 9
3回答
我使用的是Spark 1.3.1。
我正在尝试查看Python中Spark dataframe列的值。有了Spark dataframe，我可以使用df.collect()来查看数据帧的内容，但在我看来，Spark dataframe列还没有这样的方法。
例如，数据帧df包含一个名为'zip_code'的列。所以我可以做df['zip_code']，它会变成一个pyspark.sql.dataframe.Column类型，但是我找不到一种方法来查看df['zip_code']中的值。
浏览 1提问于2015-06-30得票数 43
回答已采纳
我想从nifi flow中获取数据，然后做一些事情。之后，我想再次将结果发送到nifi flow。 这是我的nifi流，使用输出端口将数据发送到spark。 ? 为了从Nifi flow中获取数据，我编写了以下函数。 def process() ={
val schema =
StructType(
Seq(
StructField(name = "ID", dataType = StringType, nullable = false),
StructField(name = "C
浏览 30提问于2020-04-09得票数 1
我天真地在本地模式下测试并发性，使用以下spark上下文
SparkSession
.builder
.appName("local-mode-spark")
.master("local[*]")
.config("spark.executor.instances", 4)
.config("spark.executor.cores", 2)
.config("spark.network.timeout", "10000001"
浏览 0提问于2019-08-25得票数 4
1回答
嗨，我在AWS glue spark工作。我从dynamodb表中获取数据并从中创建一个动态框架。我希望能够发送该表中的所有数据，记录在sqs中。我看到了另一个将动态帧转换为spark数据帧的建议。但这将是一个包含数百万条记录的表。转换为数据帧可能需要一段时间。我希望能够将动态帧中的所有记录发送到sqs队列。
下面是我的代码：
sqs = boto3.resource('sqs')
sqs_queue_url = f"https://sqs.us-east-1.amazonaws.com/{account_id}/my-stream-queue"
queue
浏览 3提问于2020-09-28得票数 1
我将一个数据帧传递给一个函数。有时是Pandas数据帧，有时是Spark数据帧。我的函数需要采取相应的行动。有没有简单的方法，比如df.isPandas()，来确定一个数据帧(作为“df”接收)是Pandas数据帧还是Spark数据帧？提前谢谢。
浏览 17提问于2019-05-14得票数 1
回答已采纳
在我当前的场景中，Nifi收集数据，然后发送给Kafka。然后，任何流媒体引擎都会消耗卡夫卡的数据，并对其进行分析。在这种情况下，我不想在Nifi和流媒体引擎之间使用卡夫卡。因此，我想将数据从Nifi直接发送到流引擎。但是，我不知道一些细节。
例如，结构化流；如果我直接将数据从Nifi发送到Spark结构化流，Spark收到了这些数据，但是Spark的节点被关闭了。火花节点中的数据会发生什么变化？( Spark结构化流是否有任何Nifi接收器？)在这种情况下，Spark结构化流的数据保证是什么？
例如，；Storm有Nifi Bolt。但是，假设Storm已经收到了Nifi的数据，但是节点被关
浏览 0提问于2019-02-14得票数 0
回答已采纳
在Apache Spark中，有没有将余弦距离的KMeans应用于Tf-IDF处理的数据帧的工作实现？
Spark，当然在ml库中有欧几里得距离实现，但对于任何其他距离度量都不是这样。
浏览 0提问于2017-06-26得票数 1
在熊猫中，这可以通过column.name来完成。
但是，当它的spark数据帧列时如何做同样的事情呢？
例如，调用程序有一个spark数据帧: spark_df
>>> spark_df.columns
['admit', 'gre', 'gpa', 'rank']
此程序调用我的函数: my_function(spark_df 'rank‘)在my_function中，我需要列的名称，即’rank‘
如果是pandas数据帧，我们可以在my_function内部使用
>>> pand
浏览 1提问于2016-09-28得票数 55
1回答
我有一个数据集存储在一个标签分隔的文本文件中。该文件如下所示：
date time temperature
2010-01-01 12:00:00 10.0000
...
其中temperature列包含以摄氏度(°C)为单位的值。我用达斯克计算日平均温度。这是我的代码：
from dask.distributed import Client
import dask.dataframe as dd
client = Client("<scheduler URL")
inputDataFrame = dd.read_table("<in
浏览 1提问于2018-10-12得票数 0
我有一个spark数据帧，经过多次转换后，它需要与其父数据帧之一连接。除非我将列'year‘重命名为'year’，否则此连接将失败。我以前也遇到过这样的行为，在6-7次转换之后，需要将数据帧与第三次转换的输出连接起来。
我不明白为什么会发生这种情况，所以我尝试了一些随机的东西，比如持久化，尝试使用spark sql API而不是pyspark，但仍然得到了同样的问题。在使用spark sql的情况下，连接在使用相同名称重命名列后工作
由于一些限制，我不能共享代码，但一般代码流如下
DF = spark.read(.......)
subset DF
df1 =
浏览 25提问于2019-08-17得票数 0
回答已采纳
我是大数据平台的新手。您能告诉我不同的方法:1)我们如何连接到spark for Spark sql? 2)我们如何连接到spark数据帧或数据集？
对于hive，我们通过unix与beeline连接，如果我们有像teradata这样的sql工具，我们必须通过安装了hive驱动的odbc连接到hive。
如何连接到spark sql和spark数据帧或数据集？
另外，如果有人能给我提供一个很好的链接或资源，让像我这样的新手理解概念、命令及其用法，easily.That将会非常有帮助。
浏览 13提问于2020-06-09得票数 0
我有两个数据帧，第一个数据帧包含实际数据(从CSV文件读取)，第二个数据帧包含一个具有多个关键字的列。即下面的Dataframe-1外观相似(我们必须搜索的地方)：
数据帧-2：
我想要的输出：
我在这里使用spark scala。我想要一个与dataframe-1中的dataframe-2完全匹配的单词。我使用了like、rlike、contains等函数，但它没有给出我想要的输出。有人知道如何在spark scala SQL中或使用dataframe的spark scala函数来开发此逻辑吗？请帮我弄一下这个。
浏览 1提问于2021-02-12得票数 0
我们已经使用Spark 1.6.1和Scala2.10编写了一些代码来通过GraphX连接和链接数据集。
如果我们在spark-shell中运行这段代码，它将在30分钟内完成，具有10个执行器，10g内存和每个执行器5个cpu。
如果我们使用spark-submit在一个fat jar中运行它，那么这个过程就会因为内存不足错误而失败，并且需要一个半小时才能到达那个点。
有没有人知道这可能是什么原因？
有谁知道SparkContext和SQLContext是如何在shell中设置的，以及我们在设置自己的上下文时是否遗漏了什么，这可能会导致spark-submit运行得如此糟糕？
浏览 18提问于2017-02-02得票数 1
回答已采纳
我是Spark新手。在我的应用程序中，我希望从许多numpy数组创建一个RDD。每个numpy数组是(10,000，5,000)。目前，我正在尝试以下操作：
rdd_list = []
for np_array in np_arrays:
pandas_df = pd.DataFrame(np_array)
spark_df = sqlContext.createDataFrame(pandas_df) ##SLOW STEP
rdd_list.append(spark_df.rdd)
big_rdd = sc.union(rdd_list)
所有的步骤都很快，除了将
浏览 0提问于2016-06-28得票数 0
以前我的理解是，一个动作会在spark应用程序中创建一个职位。但是让我们看看下面的场景，其中我只是使用.range()方法创建一个数据帧
df=spark.range(10)
因为我的spark.default.parallelism是10，所以结果数据帧是10个分区。现在我只是在数据帧上执行.show()和.count()操作
df.show()
df.count()
现在，当我检查了spark历史记录时，我可以看到.show()的3个作业和.count()的1个作业。
为什么.show()方法有3个任务？
我读过一些文章，其中.show()最终将在内部调用.take()，它将遍历
浏览 1提问于2021-03-24得票数 0
我已经创建了一个函数，它接受一行作为输入，并给出一个字符串作为输出。我计划将此函数应用于不同模式的各种数据帧。这些数据帧非常庞大，每个数据帧有数百万行，但每个数据帧都有一个定义好的模式 我想创建另一个函数，它将调用第一个函数，将函数的输出字符串与它发送到函数的行合并，并创建一个新的数据帧，该数据帧将作为第二个函数的输出。 这两个函数都将在spark-scala环境中编写。我是spark-scala的新手，不太确定如何将行组合成一个新的数据帧 def returnTranformFunctionOutput(inputDataRow: Row, TransformFrame: Array[Ro
浏览 46提问于2019-10-15得票数 0
回答已采纳
假设我的Spark集群有100G内存，在Spark计算过程中，会生成更多大小为200G的数据(新数据帧、缓存)。在这种情况下，Spark会将部分数据存储在磁盘上，还是只会使用OOM
浏览 235提问于2020-07-14得票数 0
回答已采纳
我正在构建一个spark结构的流媒体应用程序，其中我有一个具有数千行的数据帧。当我调用writeStream().format(" kafka ")时，它对kafka进行了1000次调用，以传输这些单独的行条目。 是否可以批量处理这1000条消息并对Event hub (使用KAFKA端点启用)进行一次调用？ 通过使用eventhub客户端库和调用createBatch()方法，我尝试了下面的代码。但它选择了一个字符串条目的集合，为此，我需要在dataframe上执行foreachBatch或collect才能使用该库。 EventDataBatch eventDataBa
浏览 16提问于2021-08-03得票数 0
1回答
我的用例如下:我们正在使用Spark从HDFS读取和转换数据。转换后的数据应保存在memsql中。
有没有使用Spark写入memsql的最佳实践？也就是说，在使用数据帧进行大规模(并行)写入时，确保写入性能和可伸缩性的最佳方法是什么？
首选的解决方案是:使用memsql/Spark连接器()还是DataFrame的“写”方法(与MySQL JDBC驱动程序结合使用)。
长话短说: Spark scale如何写入memsql？
浏览 4提问于2016-01-15得票数 1
3回答
将行追加到数据帧
、、
我想实现下面的一个火花，一个数据帧。我希望继续将新行添加到数据帧中，如下面的示例所示。 for(a<- value)
{
val num = a
val count = a+10
//creating a df with the above values//
val data = Seq((num.asInstanceOf[Double], count.asInstanceOf[Double]))
val row = spark.sparkContext.parallelize(
浏览 38提问于2020-01-08得票数 0
在过去，对于需要大量处理负载的作业，我会使用Scala和并行集合。
我目前正在试验Spark，发现它很有趣，但学习曲线很陡峭。我发现开发很慢，因为必须使用简化的Scala API。
在决定是否使用Spark之前，我需要确定什么？
当前Spark作业im正在尝试实现的是处理大约5 5GB的if数据。这些数据不是很大，但我正在运行这些数据的笛卡尔乘积，这会产生超过50 in的数据。但也许使用Scala并行收集器会更快，我知道从我的角度来看，实现这项工作的开发时间会更快。
那么在决定使用Spark之前，我应该考虑哪些因素呢？
浏览 0提问于2014-05-22得票数 2
我可以使用以下命令将json文件读入Pyspark中的dataframe中
spark = SparkSession.builder.appName('GetDetails').getOrCreate()
df = spark.read.json("path to json file")
但是，当我尝试将bz2(压缩的csv)读取到数据帧中时，它给我一个错误。我正在使用：
spark = SparkSession.builder.appName('GetDetails').getOrCreate()
df = spark.read.load(&#
浏览 1提问于2018-06-05得票数 7
1回答
我在读取spark数据帧时发现了这个奇怪的问题。我将数据帧重新划分为50k个分区。但是，当我读取数据帧并对其执行计数操作时，我发现当我使用spark 2.0时，底层rdd只有2143个分区。
因此，我转到保存重新分区的数据的路径，发现
hfs -ls /repartitionedData/ | wc -l
50476
因此，它在保存数据的同时创建了50k个paritions。
然而，在spark 2.0中，
val d = spark.read.parquet("repartitionedData")
d.rdd.getNumPartitions
res4: Int = 214
浏览 11提问于2017-08-11得票数 4
回答已采纳
正在尝试创建spark数据流writeStream函数的测试，如下所示：
SparkSession spark = SparkSession.builder().master("local").appName("spark
session").getOrCreate()
val lakeDF = spark.createDF(List(("hi")), List(("word", StringType, true)))
lakeDF.writeStream
.trigger(Trigger.Once)
.format
浏览 0提问于2018-07-19得票数 8
2回答
使用Spark 2.0，我发现可以将row的数据帧转换为case类的数据帧。当我尝试这样做时，我收到一条消息，声明要导入spark.implicits._。我的问题是，Intellij没有认识到这是一个有效的导入语句，我想知道它是否已经移动而消息没有更新，或者如果我的构建设置中没有正确的包，这是我的build.sbt
libraryDependencies ++= Seq(
"org.mongodb.spark" % "mongo-spark-connector_2.11" % "2.0.0-rc0",
"org.apache.s
浏览 1提问于2016-10-11得票数 32
回答已采纳
spark工作中的except()函数，用于比较两个数据帧并从第一个数据帧返回不匹配的记录。
但是，我也想跟踪字段详细信息，这是不匹配的。在spark中怎么做?？请帮帮忙
浏览 1提问于2018-06-21得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
