我正在尝试运行这样一个PySpark应用程序：
with SparkSession.builder.appName(f"Spark App").getOrCreate() as spark:
dataframe_mysql = spark.read.format('jdbc').options(
url="jdbc:mysql://.../...",
driver='com.mysql.cj.jdbc.Driver',
dbtable='my_table',
浏览 3提问于2022-08-12得票数 0
回答已采纳
1回答
我知道我们可以使用spark从现有的mysql表中读取和写入数据。但是，我们甚至可以创建mysql表并使用dataframes将数据插入其中吗？当我试图将文件加载到dataframe并试图写入不存在的表时，我面临的是空指针，exception.Following是一个错误：
org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:99)的java.lang.NullPointerException在org.apache.spar
浏览 1提问于2018-03-12得票数 2
我正在运行MySQL Docker镜像并创建数据库和表。使用以下命令加载容器：
docker run --network host -p 3306:3306 --name mysql -e MYSQL_ROOT_PASSWORD=root -d mysql/mysql-server
Spark在我的主机上运行，所以我想将Spark中的数据写入容器中运行的数据库中。
但是连接没有成功，我得到了一个异常：
com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure
浏览 1提问于2019-12-11得票数 1
1回答
如何在任务或作业完成后在控制台( Shell或submit作业)上收集这些指标。
我们使用Spark将数据从Mysql加载到Cassandra，这是相当大的(例如:200 GB和600 m行)。当任务完成时，我们要验证到底有多少行触发了进程？我们可以从Spark中获得号码，但是我们如何从火花外壳或火花提交作业中检索这个数字(“输出记录编写的”)。
示例命令从Mysql加载到Cassandra。
val pt = sqlcontext.read.format("jdbc").option("url", "jdbc:mysql://...:3306/...&
浏览 0提问于2016-04-27得票数 11
回答已采纳
1回答
我正在通过JDBC将数据从MYSQL服务器加载到Spark，但我需要在加载数据后关闭该连接。关闭连接的确切语法是什么？
df_mysql = sqlContext.read.format("jdbc").options(
url="jdbc:mysql://***/****”,
driver="com.mysql.jdbc.Driver",
dbtable="((SELECT jobid, system, FROM Jobs LIMIT 500) as T)",
user=“*****”,
password=“***
浏览 13提问于2016-09-17得票数 1
我已经注册了一个MySQL数据源，如下所示：
val driver = "com.mysql.jdbc.Driver"
val url = "jdbc:mysql://address=(protocol=tcp)(host=myhost)(port=3306)(user=)(password=)/dbname"
val jdbcDF = sqlContext.load("jdbc", Map(
"url" -> url,
"driver" -> driver,
"dbtable&#
浏览 1提问于2015-10-01得票数 1
1回答
我正在尝试将用PySpark在DataBricks中创建的数据文件加载到MySql，但它告诉我：
com.mysql.cj.jdbc.exceptions.CommunicationsException：通信链路故障
df_videojuegos.select("Nombre", "Plataforma", "Año", "Genero", "Editorial", "Ventas NA", "Ventas EU", "Ventas JP", "Ventas
浏览 6提问于2022-11-04得票数 0
回答已采纳
我有一个到MemSQL的数据加载作业(火花连接器)，由于列名的长度超过了允许的限制，它失败了。有办法解决这个问题吗？我不能更改列名，因为它们是编程生成的，我对此没有控制权。
错误信息：
Exception in thread "main" com.mysql.jdbc.exceptions.jdbc4.MySQLSyntaxErrorException: Identifier name '10000_BREAKING_BAD_IS_WAY_BETTER_THAN_THE_GAME_OF_THRONES_10000_LOWER_TOLERANCE' is too
浏览 4提问于2015-10-22得票数 0
回答已采纳
1回答
试图加载jar和外部类
、、、
在我的zeppelin-env.sh中，我以jar方式加载一个JDBC MySQL连接器，如下所示
export ZEPPELIN_JAVA_OPTS+=" -Dspark.jars=/usr/local/opt/mysql-connector-java/libexec/mysql-connector-java-5.1.32-bin.jar"
此外，我还想加载Databricks CSV包，它应该以2种(或更多)方式工作：
%dep z.load("com.databricks:spark-csv_2.10:1.2.0")
export SPARK_SU
浏览 3提问于2015-11-11得票数 0
回答已采纳
我正在使用Spark2.3，并使用jdbc从MySQL加载数据，如下所示
val dataSet:Dataset[Row] = _spark
.read
.format("jdbc")
.options(Map("url" -> jdbcUrl
,"user" -> username
,"password" -> password
,"dbtable" -> dataS
浏览 0提问于2018-12-26得票数 3
我在mysql上有一个>5 5GB的表。我想将该表作为数据帧加载到spark上，并从中创建一个拼接文件。
这是我的python函数来完成这项工作：
def import_table(tablename):
spark = SparkSession.builder.appName(tablename).getOrCreate()
df = spark.read.format('jdbc').options(
url="jdbc:mysql://mysql.host.name:3306/dbname?zeroDateTimeBehav
浏览 0提问于2017-08-08得票数 1
我有：- Hadoop -星火JobServer - SQL数据库
我已经创建了一个文件，用于从星火JobServer的本地实例访问我的SQL。为了做到这一点，我首先必须使用以下命令加载JDBC驱动程序：Class.forName("com.mysql.jdbc.Driver");。但是，当我试图在Spark JobServer上执行文件时，我会得到一个classNotFound错误：
"message": "com.mysql.jdbc.Driver",
"errorClass": "java.lang.ClassNot
浏览 1提问于2016-01-26得票数 0
回答已采纳
1回答
我现在正在学习星火，我知道如何处理特定的表格。
但是我想从数据库中加载所有的表。问题是如何做到这一点。
下面是我从特定表加载的代码：
package gt.example.spark;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import java.util.Arrays;
import java.uti
浏览 2提问于2016-03-01得票数 2
回答已采纳
2回答
我使用的是星火流2.1。我想定期刷新一些缓存的表(由spark提供的DataSource加载，比如parquet、MySQL或用户定义的数据源)。
如何刷新表格？
假设我有一张表
spark.read.format("").load().createTempView("my_table")
并且它也是由
spark.sql("cache table my_table")
下面的代码足够刷新表了吗?下一个加载表时，它将自动缓存。
spark.sql("refresh table my_table")
还是我必须用手动的方法
sp
浏览 5提问于2017-08-22得票数 18
1回答
我在客户端模式下使用python2.7和星火独立集群。
我想对mysql使用jdbc，发现我需要使用--jars参数加载jdbc，我在本地有jdbc，并设法用pyspark控制台(如 )加载它。
当我在我的ide中编写python脚本时，我没有办法加载额外的jar mysql-connector-java-5.1.26.jar并保持get
没有合适的司机
错误
在客户端模式下运行python脚本、在客户端模式上使用独立集群并引用远程主服务器时，如何加载额外的jar文件？
编辑:添加了一些代码-- #############################################
浏览 0提问于2017-08-27得票数 1
回答已采纳
1回答
我试图用连接器将一个.parquet文件加载到MemSQL数据库中。
package com.memsql.spark
import com.memsql.spark.context._
import org.apache.spark._
import org.apache.spark.sql._
import org.apache.spark.sql.types._
import com.memsql.spark.connector._
import com.mysql.jdbc._
object readParquet {
def main(args: Array[Stri
浏览 3提问于2015-11-23得票数 2
回答已采纳
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
回答已采纳
所以我有一个非常大的表，其中包含数十亿行，我尝试了Spark DataFrame应用程序接口来加载数据，下面是我的代码：
sql = "select * from mytable where day = 2016-11-25 and hour = 10"
df = sqlContext.read \
.format("jdbc") \
.option("driver", driver) \
.option("url", url) \
.option("user", user) \
浏览 1提问于2016-11-25得票数 0
来自MySQL和Kafka主题的事件数据。我们将从MySQL加载历史数据，然后切换到Kafka，以便当前的data.Mysql ID也会流入Kafka。因此，消息包含事件ID。Kafka流事件将使用spark作业进行处理，并最终将数据摄取到报表中。 问题:如何加载历史数据并实时切换到Kafka源。 1-创建临时使用者组以存储所有分区的当前偏移量和当前时间戳TS1。 2-从create date <= TS1所在的MySQL表中获取最小ID和最大ID。并通过500k记录的分区进行处理。 3-完成上述处理后，将时间戳传递到主spark作业，该作业将处理存储的偏移量中的streAming数据
浏览 18提问于2020-01-23得票数 0
我使用spark.read.format("jdbc").option("query", tmpSql)从Mysql加载一个表，我可以从数据库监视器中看到一个查询select * from (xxx) where 1=0，后来我知道这个查询用于推断Spark中的表模式。然而，当我使用spark.read.format("jdbc").option("query", tmpSql).schema(xxx)时，表模式推断查询仍然存在。既然已经指定了customSchema，为什么火花仍然需要推断表模式呢？
浏览 5提问于2021-12-09得票数 0
我有一个Dataframe，它是从mysql导入的
dataframe_mysql.show()
+----+---------+-------------------------------------------------------+
| id|accountid| xmldata|
+----+---------+-------------------------------------------------------+
|1001| 12346|<AccountSetup
浏览 2提问于2018-08-29得票数 0
2回答
我想将数据从csv作为一个批处理加载到mySql。但是我可以看到将csv中的数据插入到hive数据库的教程/逻辑。有谁能好心地帮助我实现以上使用scala的spark集成？
浏览 3提问于2017-10-27得票数 0
回答已采纳
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
我有一个从MySQL加载的spark数据集，我想将每一行与数据集中的所有其他行进行比较，并使用获得的结果创建一个新的数据集。有没有办法做到这一点？
浏览 0提问于2017-03-28得票数 1
1回答
我有一个MySQL表，并且我加载它的火花。该表包含一个具有几何类型的列。当我在spark上加载表时，具有几何类型的列将成为数据帧中的二进制类型。
我的问题是：
为什么MySQL中的几何类型在火花上变成二进制类型？有什么可替代的吗?
我需要你的帮助!
谢谢!
浏览 2提问于2020-07-24得票数 0
我正在尝试使用JDBC连接将一个4000万大表加载到spark中。显然，分区加载就是这个问题的答案。 问题是我不知道需要装载的表的模式，也不知道按哪个列进行分区。那么我如何从spark内部确定这一点呢？ val s_log = spark.read.format("jdbc").option("driver", "com.mysql.jdbc.Driver")
.option("url", "jdbc:mysql://impused?user=***&password=***")
.option("
浏览 11提问于2019-06-26得票数 1
1回答
我有一个周期性的火花-scala任务，它将数据从Hive传输到MySQL。
表的结构可以简单地看作是：
+------+------+
| id | name |
+------+------+
然后由于蜂巢表太大，所以我不得不共享MySQL表。
下面是我目前的解决方案：
准备MySQL表：
mysql> show tables;
+-------------------+
| Tables_in_test_db |
+-------------------+
| shared_0 |
| shared_1 |
| shared_2
浏览 6提问于2020-04-07得票数 0
当我试图加载带有spark数据帧的mysql表时。我收到错误消息。
Caused by: java.lang.IllegalArgumentException: MONTH
at java.util.GregorianCalendar.computeTime(GregorianCalendar.java:2648)
at java.util.Calendar.updateTime(Calendar.java:3393)
at java.util.Calendar.getTimeInMillis(Calendar.java:1782)
at com.mysql.cj.jdbc.i
浏览 2提问于2018-06-20得票数 0
假设我有两个表，table1和table2。
形成的查询类似于：select * from table1 inner join table2 on table1.id = table2.id;
现在，当我们通过这个code.How加载spark数据库连接时，我们可以提到两个表，这样就可以触发一个连接吗？
供参考的代码
spark.read().format("jdbc").option("url", "jdbc:mysql://localhost:3306/databasename")
.option("driver",
浏览 3提问于2018-09-11得票数 0
2回答
首先，我构建了scala应用程序，使用这一行代码从apache中的mysql表中读取数据。
val spark = SparkSession.builder().master("local").appName("Fuzzy Match Analysis").config("spark.sql.warehouse.dir","file:///tmp/spark-warehouse").getOrCreate()
import spark.implicits._
var df = spark.read.format("jdbc
浏览 2提问于2017-10-02得票数 1
我有一个应用程序，它从mysql和postgresql读取数据并进行数据处理。
我有以下函数来初始化spark：
def init_spark():
global sc, sqlContext, sqlCtx, sql, spark
spark = SparkSession.builder.config(
'spark.driver.extraClassPath', 'path/to/mysql-connector-java.jar'
).getOrCreate()
sc = spark.sparkConte
浏览 2提问于2020-08-13得票数 1
我是Airflow的新手，当我改变代码时，我不知道如何重新加载操作符/插件。我正在使用LocalExecutor和一个外部数据库(MySql)。我已经尝试重新启动get服务器和调度程序，但在导入dags时仍然出现相同的错误：
File "/home/ec2-user/airflow/dags/extractor.py", line 2, in <module>
from airflow.contrib.operators.emr_spark_plugin import EmrSparkOperator
ImportError: No module n
浏览 0提问于2018-07-06得票数 3
回答已采纳
我正在尝试从Mysql数据库加载RDD：
package ro.mfl.employees
import org.apache.spark.{SparkConf, SparkContext}
import java.sql.{Connection, DriverManager}
import org.apache.spark.rdd.JdbcRDD
class Loader(sc: SparkContext) {
Class.forName("com.mysql.jdbc.Driver").newInstance()
def connection(): Conn
浏览 12提问于2016-09-03得票数 3
回答已采纳
1回答
我有一个spark作业，它从mysql读取一个表，但出于某种原因，spark将int列定义为布尔值。如何在表读取期间强制数据类型？
火花会议：
spark = (SparkSession.builder
.config("spark.sql.autoBroadcastJoinThreshold", -1)
.config("spark.sql.adaptive.enabled", "true")
.config("spark.jars.packages", "mysql:mysql-connector-java:8.0.
浏览 10提问于2022-10-10得票数 1
回答已采纳
我试图使用Spark数据源API从Oracle数据库加载数据。
由于我需要通过查询加载数据，所以我使用了下面的查询，该查询是从一些在线示例中收集的：
Map<String, String> options = new HashMap<>();
options.put("driver", MYSQL_DRIVER);
options.put("user", MYSQL_USERNAME);
options.put("password", MYSQL_PWD);
options.put("url", MYSQL
浏览 8提问于2016-03-01得票数 1
我想从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
回答已采纳
为了开始工作，我使用这个创建了一个jar文件。
我的Jar文件路径是
out/artifacts/sparkProgram_jar/sparkProgram.jar
一般来说，我的spark程序从MongoDB读取一个表，使用spark的mllib转换它，并将它写入MySQL。这是我的build.sbt文件。
name := "sparkProgram"
version := "0.1"
scalaVersion := "2.12.4"
val sparkVersion = "3.0.0"
val postgresVersio
浏览 2提问于2020-08-03得票数 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
回答已采纳
1回答
因此，这段代码现在可以从MySQL读取数据并完全连接。当涉及到将数据加载到Hadoop时，我现在正在接收错误。在将文件加载到Hadoop之前，我做了一个小测试，并将文件保存到家中/下载，然后它就工作了。但是现在当我给出hdfs目录时，我得到了所有这些错误。
from pyspark.sql import SparkSession
spark = SparkSession.builder\
.appName("SqoopReplacement")\
.master("local[*]")\
.config("spark.d
浏览 2提问于2021-08-06得票数 0
1回答
默认情况下，spark_read_jdbc()将整个数据库表读入Spark。我使用了以下语法来创建这些连接。
library(sparklyr)
library(dplyr)
config <- spark_config()
config$`sparklyr.shell.driver-class-path` <- "mysql-connector-java-5.1.43/mysql-connector-java-5.1.43-bin.jar"
sc <- spark_connect(master = "local",
浏览 2提问于2017-08-01得票数 5
回答已采纳
我目前使用的项目有一些POJO文件，Spark以如下方式使用这些文件： JavaRDD<MyPojo> = ...
sqlContext.createDataFrame(rdd, MyPojo.class); 但是，我还需要使用普通的MySQL代码将Java表加载到这个POJO中。我可以使用原生SQL通过POJO加载和包装数据。但我也想知道是否允许我将这个POJO标记为Hibernate实体。 提前谢谢。
浏览 10提问于2019-03-22得票数 0
回答已采纳
我使用的是spark，我想创建查询来连接数据库中的不同表。
Apache和connection已经在使用示例：
CREATE TEMPORARY VIEW jdbcTable
USING org.apache.spark.sql.jdbc
OPTIONS (
url "jdbc:mysql://XXX.XXX.XXX.XXX/mydatabase",
driver "com.mysql.cj.jdbc.Driver",
dbtable "mydatabase.mytable1",
user "XXXX",
pa
浏览 3提问于2022-03-24得票数 0
回答已采纳
是分布式文件存储(HDFS/Cassandra/S3等)是否强制spark在群集模式下运行？若有，原因为何？
Spark是用于计算海量数据的分布式数据处理引擎。假设我在mysql中存储了大量数据，我想对这些数据进行处理。Spark从mysql读取数据，并在集群节点本身上执行内存(或磁盘)计算。我仍然不能理解为什么在集群模式下运行spark需要分布式文件存储？
浏览 3提问于2021-03-25得票数 0
0回答
我正在尝试使用spark-sql和JDBC运行一个基本的java程序。我遇到了以下错误。不知道这里出了什么问题。我读过的大多数材料都没有讨论需要做些什么来解决这个问题。
如果有人能给我推荐一些关于Spark-sql (Spark-2.1.1)的好材料，那就太好了。我计划使用spark来实现ETL，连接到MySQL和其他数据源。
线程"main“org.apache.spark.sql.AnalysisException异常:未找到表或视图：myschema.mytable；第1行pos 21；
String MYSQL_CONNECTION_URL = "jdbc:my
浏览 13提问于2017-06-09得票数 5
回答已采纳
为了将大量SQL数据加载到Spark & ML中，下面哪个选项的性能更好。
选项1:使用Spark连接器将SQLData直接加载到Spark。
选项2:使用Sqoop以csv格式将SQLData加载到HDFS，然后使用Spark从HDFS读取数据。
请建议以上哪一种方法可以很好地将大型SQL数据加载到Spark。
浏览 12提问于2015-11-18得票数 3
回答已采纳
我们已经部署并运行了OpenShift v4.0。我们在Openshift中使用Open Data Hub pods框架，其中我们有我们的jupyterhub和spark。 目标是使用spark读取一堆csv文件并将其加载到mysql中。我得到的错误是在这个踏步How to set up JDBC driver for MySQL in Jupyter notebook for pyspark?中提到的。 一种解决方案是复制spark主节点中的jar文件。但是我不能以root用户的身份访问pod。 如何在Openshift中访问pod中的root？
浏览 29提问于2020-09-03得票数 1
我们有一个应用程序，它构建在java/jax-rs上，在一个端口上侦听rest调用，我们使用tomcat来部署我们的应用程序，api侦听这个post调用，在一些名义上的过滤之后将数据转储到mysql中，然后一个定时器(java.util.Timer)线程每秒查询数据并处理它，并将其存储到另一个数据库中。现在，由于传入数据的频率/数据量非常大，mysql有时会阻塞。我们正在考虑采用主主复制(因为加载是在插入时进行的，所以主从复制不会有任何帮助)，这可能会工作一段时间，但运行的算法也非常占用cpu。
因此，我们正在考虑一种永久的替代方案，它将spark引入到场景中，我们已经有了一个正在运行的hot
浏览 3提问于2019-02-01得票数 0
我们确实遇到了表查找的问题。我们需要同时从hive和spark (使用thrift服务器)访问表格。然而，我们的问题是在spark上运行spark和thrift服务器导致一个表的查找。
我们在Amazon AWS EMR集群上运行，其中包含Hive、Spark和thrift Server2。
我们希望使用hive更新s3存储，并定期在后台将这些聚合数据加载到spark中。同时，Spark总是加载了thrift服务器，并从s3加载了相同的数据，以便对这些数据进行实时聚合。Spark不需要对此数据的写访问权限。
问题是在hive上运行周期性的所有数据加载任务会导致作业冻结。
我们认为元存储可能被s
浏览 2提问于2015-12-22得票数 0
How to load a parquet file into vertica database using spark???
链接()
我试着使用上面的链接将数据框(拼图文件)加载到mysql中，它起作用了。但是当我试图将它加载到vertica数据库时，下面的错误是I am facing.The，错误是因为vertica db不支持数据帧(拼图文件)中的数据类型(字符串)。我不想对列进行类型转换，因为这将是一个性能问题。我们希望加载大约2.8亿行。你能建议一下把数据加载到vertica数据库的最好方法吗？
Exception in thread “main” java.sql.SQLSynt
浏览 4提问于2015-10-30得票数 2
1回答
目标:构建一个小型的ETL框架，以获取一个巨大的CSV并将其转储到RDB(比如MySQL)。
我们正在考虑的当前方法是使用spark将csv加载到一个dataframe中，并将其持久化，然后使用apache之类的框架，然后将其加载到mySQL中。
需要就应坚持的格式和方法本身提出建议。
编辑: CSV将有大约5000万行和50-100列.由于我们的任务在转储到RDB之前需要进行大量的转换，所以我们认为使用spark是个好主意。
浏览 3提问于2016-01-18得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
