我正在尝试设置一条从mysql binlogs到redshift的数据管道。我正在从Mysql binlogs(使用一个工具- cannedbeer，来自mypipe的fork )向kafka写入数据，然后使用spark streaming将这些消息写入到使用spark redshift connector.The的Redshift中。我面临的问题是，由于Dstream .Can的foreachRDD方法(写入Redshift的副作用)中的作业失败，相同的消息被多次写入到redshift.Is这是因为这个问题以及如何提前解决it.Thanks。
浏览 1提问于2015-11-12得票数 1
目前，我正在使用JDBC驱动程序将数据从MySQL数据库导入到spark中，使用以下命令：
dataframe_mysql = sqlctx
.read
.format("jdbc")
.option("url", "jdbc:mysql://<IP-ADDRESS>:3306/<DATABASE>")
.option("driver", "com.mysql.jdbc.Driver")
.option("dbtable", "
浏览 4提问于2017-08-11得票数 3
1回答
我正在从spark应用程序连接到一个MySql DB (星火-2.0.0-bin-hadoop2.7)。当连接时，我得到以下错误。
这是我的代码：
def main(args: Array[String]) {
val conf = new SparkConf()
.setAppName("LoadMySql")
.setMaster("local[*]")
.set("spark.sql.warehouse.dir", "file:///C:/temp")
val spark = Sp
浏览 1提问于2018-09-25得票数 0
1回答
我试图从Azure事件中心读取数据，并以火花流模式将此数据存储到Mysql表中。
下面是我的电火花代码
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
import json
from datetime import datetime as dt
from pyspark.sql import DataFrameWriter
try:
session = SparkSession.builder.master("lo
浏览 10提问于2021-12-22得票数 1
回答已采纳
是分布式文件存储(HDFS/Cassandra/S3等)是否强制spark在群集模式下运行？若有，原因为何？
Spark是用于计算海量数据的分布式数据处理引擎。假设我在mysql中存储了大量数据，我想对这些数据进行处理。Spark从mysql读取数据，并在集群节点本身上执行内存(或磁盘)计算。我仍然不能理解为什么在集群模式下运行spark需要分布式文件存储？
浏览 3提问于2021-03-25得票数 0
在SparkSQL中可以加入mysql和Oracle数据库中的数据吗？我试图加入他们，但在SPARK_CLASSPATH中设置多个jars (用于mysql和Oracle的jdbc驱动程序)时遇到了一些问题。这是我的代码：
import os
import sys
os.environ['SPARK_HOME']="/home/x/spark-1.5.2"
sys.path.append("/home/x/spark-1.5.2/python/")
try:
from pyspark import SparkContext, Spar
浏览 2提问于2015-12-30得票数 1
当我为表同步运行spark应用程序时，错误消息如下所示：
19/10/16 01:37:40 ERROR Executor: Exception in task 0.0 in stage 3.0 (TID 51)
com.mysql.cj.jdbc.exceptions.CommunicationsException: Communications link failure
The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packet
浏览 3提问于2019-10-16得票数 0
回答已采纳
在Scala中用MySQL将数据火花流存储到foreachRDD()中
请有人给我一个函数示例，说明如何使用Scala中的MySQL foreachRDD()来保存流到foreachRDD()的Spark。我有下面的代码，但它不起作用。我只需要一个简单的例子，而不是正法或理论。
谢谢!
package examples
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark._
import org.apache.spark.storage.StorageLevel
import org.apache.
浏览 1提问于2017-05-20得票数 4
回答已采纳
1回答
我的用例如下:我们正在使用Spark从HDFS读取和转换数据。转换后的数据应保存在memsql中。
有没有使用Spark写入memsql的最佳实践？也就是说，在使用数据帧进行大规模(并行)写入时，确保写入性能和可伸缩性的最佳方法是什么？
首选的解决方案是:使用memsql/Spark连接器()还是DataFrame的“写”方法(与MySQL JDBC驱动程序结合使用)。
长话短说: Spark scale如何写入memsql？
浏览 4提问于2016-01-15得票数 1
我想用Python (PySpark)实现从Kafka源代码到MariaDB (PySpark)的Spark结构化流(Spark2.4.x)。
我想使用流式星火数据，而不是静态或潘达斯的数据。
似乎必须使用foreach或foreachBatch，因为根据，流数据没有可能的数据库接收器。
以下是我的尝试：
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql.types import StructField, StructType, StringType, DoubleTy
浏览 3提问于2019-11-08得票数 7
回答已采纳
嗯，我使用的是PySpark，我有一个Spark dataframe，我使用它将数据插入到mysql表中。
url = "jdbc:mysql://hostname/myDB?user=xyz&password=pwd"
df.write.jdbc(url=url, table="myTable", mode="append")
我希望通过列值和特定数字的总和来更新列值(不在主键中)。
我尝试过不同的模式(追加、覆盖) DataFrameWriter.jdbc()函数。
我的问题是，我们如何像在mysql中使用ON DUPLICATE K
浏览 4提问于2015-09-16得票数 12
我试图弄清楚如何使用新的DataFrameWriter将数据写回JDBC数据库。我似乎找不到这方面的任何文档，尽管看一下源代码，它似乎是可能的。
下面是我正在尝试的一个简单的例子：
sqlContext.read.format("jdbc").options(Map(
"url" -> "jdbc:mysql://localhost/foo", "dbtable" -> "foo.bar")
).select("some_column", "another_column
浏览 0提问于2015-09-16得票数 7
1回答
目标:构建一个小型的ETL框架，以获取一个巨大的CSV并将其转储到RDB(比如MySQL)。
我们正在考虑的当前方法是使用spark将csv加载到一个dataframe中，并将其持久化，然后使用apache之类的框架，然后将其加载到mySQL中。
需要就应坚持的格式和方法本身提出建议。
编辑: CSV将有大约5000万行和50-100列.由于我们的任务在转储到RDB之前需要进行大量的转换，所以我们认为使用spark是个好主意。
浏览 3提问于2016-01-18得票数 2
回答已采纳
2回答
我是一个使用Spark的新手。我遵循了DataBricks：的在线指南
我可以成功地获得到MySQL实例的连接并从中读取。但是，我一直从Spark中获得NoTableFound或NoDatabaseFound错误的变体。下面是我的整个测试类的样子：
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.SQLContext;
import org.a
浏览 4提问于2017-05-01得票数 0
回答已采纳
2回答
我有一个DSE集群，集群中的每个节点都在运行spark和Cassandra。
当我从Cassandra加载数据到spark rdd并在Rdd上执行一些操作时，我知道数据将分布到多个节点中。在我的例子中，我想把这些rdds从每个节点直接写到它的本地Cassandra dB表中，有没有办法做到这一点。
如果我执行普通的rdd收集，来自spark节点的所有数据将被合并，并返回到带有驱动程序的节点。我不希望发生这种情况，因为从节点返回到驱动程序节点的数据可能需要很长时间，我希望数据直接保存到本地节点，以避免数据跨spark节点移动。
浏览 3提问于2020-10-20得票数 1
最终，我想从Vertica DB获取数据到Spark，训练机器学习模型，进行预测，并将这些预测存储到另一个Vertica DB中。
当前的问题是识别流的最后一部分中的瓶颈:从Spark存储Vertica DB中的值。在Vertica DB中存储63k行数据大约需要38分钟。相比之下，当我将相同的数据从Spark传输到MySQL数据库时，需要10秒。
我不知道为什么有这么大的差别。
我有名为VerticaContext和MySQLContext的类，分别用于Vertica和MySQL连接。两个类都使用来使用jdbc格式读取条目。
df = self._sqlContext.read.format
浏览 3提问于2016-04-19得票数 0
1回答
首先，我在Amazon上使用Spark1.5.2，在postgres数据库中使用Amazon。
我目前正致力于让Spark将其结果写入postgres数据库，我正在使用Spark的jdbc进行此操作。当它正常工作时，我担心的是，如果我试图在Spark运行时使用psql连接到db，然后尝试从Spark正在写入的表中读取，它就会告诉我还没有写行。这种情况一直持续到火花结束，然后所有的行突然出现在数据库中。
这让我很担心，因为最终的设置应该是定期运行Spark脚本，并将更多的数据写入数据库，而网站应该使用该数据库。这是否意味着我无法在Spark运行时查询数据库？
你猜这里的问题是什么？我认为这可能是
浏览 1提问于2016-01-21得票数 0
回答已采纳
1回答
我正在从container1中检索两个文件，转换它们，并在写入到Azure相同存储帐户中的container2之前进行合并。我正在安装container1，在编写之前打开和安装countainer2。
我写拼花的代码
spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
df_spark.coalesce(1).write.option("header",True) \
.partitionBy('ZMTART') \
浏览 8提问于2022-11-22得票数 0
我从设置hive.metastore.uris的metastore启动spark应用程序中读取表统计数据。但是，我需要将数据写入另一个配置单元。 我已经尝试清理活动会话和默认会话，使用新的metastore uri构建另一个会话，但spark继续尝试写入第一个hive的表。 val spark = SparkSession.builder()
.appName(appName)
.enableHiveSupport()
.config("hive.metastore.uris", FIRST_METASTORE)
浏览 9提问于2019-04-11得票数 0
当试图将JDBC加载到Spark中时，我遇到了非常奇怪的问题。
我尝试了几个星团-纱线，独立的集群和伪分布式模式在我的笔记本电脑。它在星火1.3.0和1.3.1上都是可复制的。这个问题既发生在spark-shell中，也发生在用spark-submit执行代码时。我尝试过MySQL &MySQL驱动程序，但没有成功。
考虑以下样本：
val driver = "com.mysql.jdbc.Driver"
val url = "jdbc:mysql://localhost:3306/test"
val t1 = {
sqlContext.load(
浏览 2提问于2015-04-28得票数 13
回答已采纳
1回答
我目前正在尝试通过Spark SQL将一个非常大的MySQL表的内容批量迁移到一个parquet文件中。但是当我这样做的时候，我很快就会耗尽内存，即使是在驱动程序的内存限制设置得更高的时候(我在本地模式下使用spark )。示例代码：
Dataset<Row> ds = spark.read()
.format("jdbc")
.option("url", url)
.option("driver", "com.mysql.jdbc.Driver")
.option("dbtab
浏览 12提问于2016-08-20得票数 7
回答已采纳
使用Spark1.4.0，我尝试使用DataFrame ()将来自Spark MemSQL的数据插入到MemSQL数据库中(应该与MySQL数据库进行完全类似的交互)。但是，我一直得到一个运行时TableAlreadyExists异常。
首先，我创建如下的MemSQL表：
CREATE TABLE IF NOT EXISTS table1 (id INT AUTO_INCREMENT PRIMARY KEY, val INT);
然后，我在Spark中创建一个简单的dataframe，并尝试插入到MemSQL中，如下所示：
val df = sc.parallelize(Array(123,2
浏览 0提问于2015-10-02得票数 12
回答已采纳
我想把输出数据导入mysql数据库，但是发生以下错误，我不会将数组转换成所需的字符串类型，能帮我吗？
val Array(trainingData, testData) = msgDF.randomSplit(Array(0.9, 0.1))
val pipeline = new Pipeline().setStages(Array(labelIndexer, word2Vec, mlpc, labelConverter))
val model = pipeline.fit(trainingData)
val predictionResultDF = model.tr
浏览 0提问于2018-05-06得票数 2
回答已采纳
1回答
我正在一个Scala + Spark项目中工作，在该项目中，我们将数据从文件加载到PostgreSQL中。它在独立模式下使用jdbc.write在本地运行良好，测试数据很小。
但是由于生产数据是巨大的，我想使用一个集群，每个执行者拥有多个工作人员和一个逻辑处理器核心。
考虑到这一点，我如何在集群中的所有可用核之间划分数据？
谢谢!
PS:使用Scala2.13.9和Spark3.3.0
浏览 4提问于2022-10-12得票数 1
回答已采纳
2回答
我正在学习PySpark，并编写了一个简单的脚本，该脚本从我的HDFS目录中加载一些JSON文件，将每个JSON文件作为python字典加载(使用json.loads() )，然后为每个对象提取一些字段。
相关信息存储在中，我希望将这些数据插入到MySQL表中(我是在本地创建的)。
但是，当我运行这个时，我的连接URL会出现一个错误。
上面写着"java.lang.RuntimeException: 1.5失败：。“
在这一点上：
jdbc:mysql://localhost:3306/bigdata?user=root&password=pwd
^
数据库名为"
浏览 3提问于2017-05-02得票数 0
回答已采纳
我有一个类似下面的pyspark脚本。在这个脚本中，我遍历表名的input文件并执行代码。
现在，我想在每次迭代函数mysql_spark时分别收集日志。
例如：
input file
table1
table2
table3
现在，当我执行pyspark脚本时，我将所有三个表的日志保存在一个文件中。
What I want is 3 separate log files 1 for each table
Pyspark脚本：
#!/usr/bin/env python
import sys
from pyspark import SparkContext, SparkConf
from py
浏览 1提问于2017-07-29得票数 0
回答已采纳
在下面的场景中，我搞不懂Spark会建立多少到数据库的连接：
假设我有一个Spark程序，它只在一个具有一个执行器的工作节点上运行，一个数据帧中的分区数是10，我想将这个数据帧写入Teradata。由于并行度是10，但是executor只有1，那么在保存数据时会建立10个连接，还是只有1个连接？
浏览 3提问于2020-11-05得票数 1
我正在获取mongoDB文档，然后在处理之后，我想使用Bson.Document库将其存储到Hbase中
将流媒体方法从spark kafkastreaming改为结构化流媒体，因此早期使用kafkaUtils的方法产生了DstreamDocument
在结构化流媒体中，我得到了DatasetDocument
scala> val stream = spark.readStream.format("kafka").option("kafka.bootstrap.servers","brokerList").option("subsc
浏览 15提问于2019-11-09得票数 2
我有这样的场景，我将接收由我的spark流程序处理的流数据，并且每个间隔的输出被附加到我现有的cassandra表中。
目前，我的spark流程序将生成一个数据帧，我需要将其保存在我的cassandra表中。我目前面临的问题是，当我使用下面的命令时，我无法将数据/行附加到现有的cassandra表中
dff.write.format("org.apache.spark.sql.cassandra").options(Map("table" -> "xxx", "yyy" -> "retail")).sa
浏览 0提问于2016-02-11得票数 5
我想从mysql获得数据到Spark (scala)，但当数据发生时会出错
com.mysql.cj.jdbc.exceptions.CommunicationsException:通信链路故障
这是我的密码：
val sqlcontext = new org.apache.spark.sql.SQLContext(sc)
val cataDF= sqlcontext.read.format("jdbc").option("url", "jdbc:mysql://127.0.0.1:3360/crawldb").option("
浏览 1提问于2019-07-24得票数 0
回答已采纳
我正在编写一个Spark结构的流媒体应用程序，其中使用Spark处理的数据需要沉没到弹性搜索。
这是我的开发环境，因此我有一个独立的Elastic search。
我尝试了以下两种方法将DataSet中的数据汇聚到ES中。
1.ds.writeStream().format("org.elasticsearch.spark.sql").start("spark/orders"); 2.ds.writeStream().format("es").start("spark/orders");
在这两种情况下，我都会得到以下错误：
由以
浏览 1提问于2017-09-26得票数 3
1回答
我正在探索从Spark日志中为Spark程序获取数据谱系信息的方法。
我正在寻找像kafka主题或表Spark程序读取或写入的信息，以便我们可以获得该信息运行时，并建立端到端的数据流移动。有没有人探索过这种框架。
当我有Info日志级别设置时，我可以获得有关输入kafka读取和表写入数据的信息，但是，如果数据被发送到Kafka主题或输入表读取，我无法获得信息。
感谢您的帮助。
谢谢&致以问候。
浏览 3提问于2017-11-13得票数 0
我感兴趣的是如何插入例如10 GB的.csv文件到MySQL数据库。我使用pandas和pyspark来读取csv文件，然后添加csv头到列表中(如果csv文件没有头，我使用spark添加它)。然后我解析列表并替换MySQL插入代码的字符--> mydb.cursor().execute("CREATE TABLE " + table_name + " (id INT NOT NULL AUTO_INCREMENT," + column_names + ", PRIMARY KEY (id))") 然后，我将不带标题行的整行添加到列表中，
浏览 20提问于2021-07-11得票数 0
我有一个从远程存储读取数据的场景：
df = spark.read.load("abfss://mycontainer@mystorageacct.dfs.core.windows.net/mydata.csv")
这是一个包含几个GBs的小数据集，加载大约需要1到2分钟。
同样的笔记本电脑每天都是手动运行的，我想做一些小的优化。
cache()和persist()似乎没有帮助，因为在会话结束时数据将是uncached\unpersisted？
将数据写入集群上的本地存储并从中读取数据是一种ok模式吗？
localfile = '/X/myfile.parquet
浏览 3提问于2022-10-26得票数 0
回答已采纳
当数据被插入到表中时，我正在尝试立即读取cassandra表。该表将时间戳作为主键之一(而不是分区键)。我们有一个spark任务，每15秒读一次kafka，写一次cassandra。当spark作业开始插入数据时，服务器组件几乎立即从cassandra读取数据。由于插入到cassandra的数据非常庞大，我们正在分页读取数据。在阅读页面时，我们观察到很少有记录被跳过并到达最后一条记录。
但是，当我们在所有已插入的数据上运行按页读取数据的相同逻辑时，它工作得很好(没有跳过记录)。在cassandra中插入数据时，有没有办法读取分页的数据？
浏览 2提问于2016-05-13得票数 0
我有一个火花作业，它在第一次运行时在5分钟内运行，然后需要几分钟..more，而在随后的运行中需要20-30分钟。我正在读取一个拼花文件一次，然后创建数据格式并以.json格式编写。在代码中的任何地方，我都没有使用缓存()、持久化()或未持久化()。这是本地实例。有什么问题吗？
配置参数
val spark = SparkSession
.builder()
.appName("example")
.config("spark.sql.warehouse.dir", warehouseLocation)
.config("spark
浏览 2提问于2017-07-28得票数 0
我们通过EMR使用Apache Spark(2.4.5)作业，它读取存储桶前缀{ S3 }/{ prefix }/*.json，执行一些数据消息，然后在覆盖模式下通过Spark作业save()将其重写回相同的{bucket}/{prefix}。我的问题是，如果Spark作业在将数据重写到S3前缀分区时失败，那么我们是否有办法以任何原子/事务方式恢复该前缀分区中的数据。spark/EMR/S3中的任何/所有这些都支持它吗？
浏览 11提问于2020-10-02得票数 0
我设计了一个简单的工作，可以从MySQL读取数据，并将其保存在使用Spark的Elasticsearch中。
以下是代码：
JavaSparkContext sc = new JavaSparkContext(
new SparkConf().setAppName("MySQLtoEs")
.set("es.index.auto.create", "true")
.set("es.nodes", "127.0.0.1:9200")
浏览 0提问于2015-09-19得票数 7
回答已采纳
1回答
我在用scala构建一个火花罐时遇到了问题。这是一件非常简单的事情，我想通过JDBC编程地访问mysql服务器，并将它加载到星星之火数据帧中。我可以让它在火花壳中工作，但我不能打包一个与火花提交一起工作的罐子。它将打包，但在运行时，将失败
Exception in thread "main" java.sql.SQLException: No suitable driver found for jdbc:mysql://localhost:3310/100million
我的火花提交命令是
./bin/spark-submit ~/path/to/scala/project/t
浏览 4提问于2015-10-21得票数 1
回答已采纳
我试图开发几个数据管道使用Apache气流与预定的火花作业。
对于这些管道之一，我试图将数据从PySpark DataFrame写入MySQL，并且一直遇到一些问题。这只是我的代码现在看起来的样子，但是我确实想在将来添加更多的转换，
df_tsv = spark.read.csv(tsv_file, sep=r'\t', header=True)
df_tsv.write.jdbc(url=mysql_url, table=mysql_table, mode="append", properties={"user":mysql_user,
浏览 2提问于2021-10-28得票数 1
回答已采纳
我在MySQL中有一个test表，其id和名称如下：
+----+-------+
| id | name |
+----+-------+
| 1 | Name1 |
+----+-------+
| 2 | Name2 |
+----+-------+
| 3 | Name3 |
+----+-------+
我使用Spark读取这些数据(使用JDBC)并修改数据，如下所示
Dataset<Row> modified = sparkSession.sql("select id, concat(name,' - new') as name from
浏览 0提问于2017-01-26得票数 8
回答已采纳
我试图使用Spark、DataFrames和JDBC连接在MySql上插入和更新一些数据。
我已经成功地使用SaveMode.Append插入了新数据。是否有方法从Spark更新MySql表中已经存在的数据？
我要插入的代码是：
myDataFrame.write.mode(SaveMode.Append).jdbc(JDBCurl,mySqlTable,connectionProperties)
如果我更改为SaveMode.Overwrite，它将删除完整的表并创建一个新的表，我正在寻找类似于MySql中可用的“关于重复键更新”的内容
浏览 11提问于2016-02-25得票数 32
回答已采纳
1回答
嗨，我们有spark集群，在spark作业执行期间，当我将中间数据写入spark.local.dir位置时，我得到了sparkoutofmemory，但是当我看到他们的内存是执行器未使用的两倍以上时，所以我们可以将数据存储到内存本身吗？
下面是异常详细信息
Job aborted due to stage failure: Task 134555 in stage 32.0 failed 4 times, most recent failure: Lost task 134555.3 in stage 32.0 (TID 151065, <<some worker node IP&
浏览 1提问于2020-12-29得票数 1
我有一个spark应用程序，它将处理后的输出转储到S3存储桶中。我的输出数据帧在pyspark中的分区数是300。我有列"id“(大约250个唯一值)，我用它来用partitionBy写文件。因此，我得到了大约75K个文件，总和为11 to。
在我的spark应用程序中，将此数据写入S3的所有阶段都已完成，但从_temp文件夹中移动文件仍在进行中。这是非常缓慢的，按照目前的速度，这将需要一天以上的时间。这里的根本原因是什么?如何优化它？
浏览 1提问于2020-06-05得票数 0
我想使用我的天蓝色数据湖Gen2上的蜂巢Metastore中的达美湖表作为我公司湖房的基础。
以前，我使用了“常规”蜂巢目录表。我会将数据从拼图加载到火花数据，并使用df.CreateOrReplaceTempView("TableName")创建一个临时表，这样我就可以使用spark或% SQL魔术来执行ETL了。这样做之后，我可以在spark.sql或%%sql上使用TableName。当我做完了，我会write我的桌子到蜂巢亚稳态。
但是，如果我不想执行这个saveAsTable操作并写入我的数据湖，该怎么办？使用SQL执行ETL的最佳方法是什么？
我知道我可以通过多种方
浏览 6提问于2022-09-08得票数 0
回答已采纳
我试图使用Pyspark连接插入到现有的Mysql表中，但是我得到了以下错误
File "<stdin>", line 1, in <module>
File "/usr/hdp/current/spark2-client/python/pyspark/sql/context.py", line 384, in sql
return self.sparkSession.sql(sqlQuery)
File "/usr/hdp/current/spark2-client/python/pyspark/sql/se
浏览 0提问于2020-07-17得票数 0
回答已采纳
1回答
我运行pyspark2在spark 2.0.0上
考虑到加载到Dataframe中的稳定和常量数据集，我确实按特定列删除了重复数据：
vw_ticket = read_csv(...)
vw_ticket_in_booking = vw_ticket.dropDuplicates(['BOOKING_TICKET_ID'])
vw_ticket_in_booking.createOrReplaceTempView('vw_ticket_in_booking')
然后
spark.sql('select count(distinct(booking_id)
浏览 2提问于2017-08-31得票数 0
回答已采纳
我正在写一份spark工作，根据学生日期过滤最新的学生记录。但当我尝试使用数十万条记录时，它工作得很好。但是，当我使用大量记录运行它时，我的sparkjob返回下面的错误。
我猜这个错误是因为我从表中加载了所有数据并将int放入了RDD中。因为我的表包含大约420万条记录。如果是这样的话，有没有更好的方法来有效地加载这些数据并成功地继续我的操作？
请任何人帮我解决这个问题
WARN TaskSetManager: Lost task 0.0 in stage 1.0 (TID 1, 10.10.10.10): java.lang.OutOfMemoryError: GC overhead li
浏览 22提问于2017-03-10得票数 1
我们有大量的服务器数据存储在S3中(很快将以Parquet格式存储)。数据需要一些转换，因此不能直接从S3复制。我将使用Spark来访问数据，但我想知道的是，如果我可以跳过一个步骤，运行一个查询来提取/转换数据，然后直接将其复制到Redshift，那么是否可以代替用Spark操作它，将其写回S3，然后复制到Redshift？
浏览 1提问于2016-04-14得票数 6
回答已采纳
我有以下代码：
Dataset<Row> rows = sparkContext.sql ("select from hive tables with multiple joins");
rows.saveAsTable(writing to another external table in hive immediately);
1)在上述情况下，当调用saveAsTable()时，spark会将整个数据集加载到内存中吗？
1.1)如果是，那么当这个查询实际上可以返回无法放入内存的大量数据时，我们该如何处理这种情况？
2)当服务器崩溃，spark开始执行saveA
浏览 31提问于2020-06-04得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
