Spark JDBC仅返回包含列名的dataframe
Spark JDBC是Spark提供的一种用于与关系型数据库进行交互的工具。它允许用户通过Spark来读取和写入关系型数据库中的数据。
Spark JDBC可以通过连接数据库并执行SQL查询来获取数据。当使用Spark JDBC执行查询时，返回的DataFrame只包含列名，而不包含实际的数据。这是因为Spark JDBC是一种延迟加载的机制，它只在需要时才从数据库中获取数据。
返回的DataFrame包含了查询结果的列名信息，这使得用户可以方便地了解查询结果的结构。用户可以通过DataFrame的操作来进一步处理和转换数据，例如过滤、聚合、排序等。
Spark JDBC的优势在于它能够与各种关系型数据库进行无缝集成，包括MySQL、Oracle、SQL Server等。它提供了高性能的数据读取和写入能力，可以处理大规模的数据集。同时，Spark JDBC还支持并行查询和分布式处理，可以利用Spark的分布式计算能力来加速数据处理过程。
Spark JDBC的应用场景非常广泛。例如，当需要从关系型数据库中读取大量数据进行分析时，可以使用Spark JDBC来高效地获取数据并进行处理。另外，当需要将Spark处理的结果写入关系型数据库时，也可以使用Spark JDBC来实现数据的写入操作。
腾讯云提供了一系列与Spark JDBC相关的产品和服务。例如，腾讯云的云数据库MySQL和云数据库SQL Server可以与Spark JDBC无缝集成，用户可以通过Spark JDBC来读取和写入腾讯云的数据库。此外，腾讯云还提供了弹性MapReduce（EMR）服务，用户可以在EMR上使用Spark来进行大数据处理，并与腾讯云的数据库进行交互。
更多关于腾讯云的Spark JDBC相关产品和服务的详细信息，您可以访问以下链接：
相关·内容
我试图开发几个数据管道使用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
回答已采纳
我可以将数据帧记录加载到Oracle数据库中，没有任何错误，但我想知道如何将Df中的列映射到Oracle 我想以JDBC write格式将Employee_id字段从DataFrame映射到oracle中的ID列。 列映射： df ---------- Oracle
Employee_id ---------- ID
targetdf=spark.sql(final_sql)
targetdf.show()
targetdf.write \
.format("jdbc") \
.option("url", "
浏览 21提问于2020-11-27得票数 0
和spark.read.format ..。可以以非编程方式添加自定义模式，如下所示：
val df = sqlContext
.read()
.format("jdbc")
.option("url", "jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true")
.option("user", "root")
.option("password", &#
浏览 0提问于2018-06-04得票数 3
回答已采纳
我正在尝试使用spark.read.format读取一个Oracle表，它对所有表都很有效，除了少数表中有任何数据类型为ROWID的列。
下面是我的代码
var df = spark.read.format("jdbc").
option("url", url).
option("driver", driver).
option("dbtable", dbTable).load()
println(df.first)
我得到下面的错误
18/09/08 11:38:17 WARN s
浏览 42提问于2018-09-09得票数 3
我正在尝试使用Simba JDBC驱动程序(bigquery jdbc 4.2)从apache spark 2.2.0中查询bigquery
这是我的代码：
import spark.implicits._
import org.apache.spark.sql.SQLContext
val sqlcontext = new org.apache.spark.sql.SQLContext(sc)
val pushdown_query = "(select * from mydb.mytable) AS T"
val dataframe_bq = sqlcontext.read.
浏览 3提问于2017-10-31得票数 2
我在Scala中使用Spark。在将数据加载到Spark Dataframe之后，我想要访问Dataframe的每个单元格来执行一些计算。代码如下： val spark = SparkSession
.builder
.master("local[4]")
.config("spark.executor.memory", "8g")
.config("spark.executor.cores", 4)
.config("spark.task.cpus",1)
.appName("Spark SQL bas
浏览 23提问于2021-09-19得票数 1
在“火花:最终指南”中，它说：
如果需要引用特定DataFrame的列，则可以在特定的DataFrame上使用col方法。
例如(在Python/Pyspark中)：
df.col("count")
但是，当我在包含列count的dataframe上运行后的代码时，就会得到错误'DataFrame' object has no attribute 'col'。如果我尝试column，我会得到一个类似的错误。
这本书是错的，还是我该怎么做呢？
我上的是星火2.3.1。dataframe是用以下内容创建的：
df = spark.read.f
浏览 2提问于2018-08-12得票数 9
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
我想尝试使用Spark从Teradata读取数据。
我有一个来自一个表的Teradata视图，其中包含列(Order_id)和标题( "ORDER ID")列名。
val Query = "(select order_id from table sample 100) ST ";
var reader = sparkSession.read
.format("jdbc")
.option("url", "jdbc:teradata://xx/charset=xx, DBS_PORT=xx
浏览 1提问于2020-02-06得票数 1
1回答
mysql数据源用于将数据加载到Spark中，在string中有一个包含JSON的列。
// JDBC Connection and load table in Dataframe
val verDf = spark.read.format("jdbc").option("driver", driver).option("url", url).option("dbtable", verticesTable).option("user", user).option("password", pass).
浏览 2提问于2021-01-20得票数 0
回答已采纳
1回答
我想作为一个数据集从黑拉拉表中获得子查询。
这样的代码：
String subQuery = "(select to_timestamp(unix_timestamp(now())) as ts from my_table) t"
Dataset<Row> ds = spark.read().jdbc(myImpalaUrl, subQuery, prop);
但结果是错误：
Caused by: java.sql.SQLDataException: [Cloudera][JDBC](10140) Error converting value to Timestam
浏览 0提问于2018-10-22得票数 0
回答已采纳
我尝试使用pyspark和jdbc将数据插入到mariadb中，但似乎pyspark没有生成正确的SQL，我的Spark版本是2.1.0，我没有这个问题，直到集群管理器将Spark从1.6.1更新到2.1.0，这是我的python代码
from pyspark.sql import Row, SparkSession as SS
if __name__ == "__main__":
spark = SS.builder.appName("boot_count").getOrCreate()
sc = spark.SparkContext
浏览 3提问于2017-01-07得票数 0
我正在使用Simba技术公司的JDBC驱动程序来连接Google云扳手。它正在按预期的方式在Java.sql中运行。当我试图将simba驱动程序与Spark的JDBC读取器一起使用时，为了将查询输出读取为DataFrame，但是它提供了错误的输出。
这是我的扳手表
UserID UserName
1 Vaijnath
2 Ganesh
3 Rahul
MetaData: UserID(字符串)
UserName(字符串)
我正在执行查询，如:从用户中选择*
当我在Java中使用Simba驱动程序时，这个查询会获取正确的数据，但是当我将它与Spark的JDBC读取器一起使用时，它将
浏览 0提问于2017-06-29得票数 3
回答已采纳
使用Spark的JDBC数据源访问Postgresql时，属性partitionColumn用作不区分大小写的属性，并引发异常
CREATE TABLE data USING org.apache.spark.sql.jdbc OPTIONS (url 'jdbc:postgresql://localhost:5432/postgres', dbtable 'public.datios', lowerBound '0', upperBound '10', partitionColumn 'COD_PERSON',
浏览 0提问于2019-08-02得票数 0
当执行以下代码段时：
...
stream
.map(_.value())
.flatMap(MyParser.parse(_))
.foreachRDD(rdd => {
val spark = SparkSession.builder.config(rdd.sparkContext.getConf).getOrCreate()
import spark.implicits._
val dataFrame = rdd.toDF();
val countsDf = dataFrame.g
浏览 3提问于2016-08-28得票数 3
回答已采纳
当我为表同步运行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
回答已采纳
我可以在Azure Databricks中使用PySpark执行一个简单的SQL语句，但是我想执行一个存储过程。下面是我尝试过的PySpark代码。
#initialize pyspark
import findspark
findspark.init('C:\Spark\spark-2.4.5-bin-hadoop2.7')
#import required modules
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession
from pyspark.sql impo
浏览 5提问于2020-02-23得票数 5
我试图使用Spark的JDBC访问存储在远程集群上的表(ORC格式)：
val jdbcDF = spark.read
.format("jdbc")
.option("url", url)
.option("dbtable", "metrics")
.option("user", user)
.option("password", password)
.load()
然而，不管我做什么，我总是会犯这样的错误：
由: org.
浏览 5提问于2017-05-03得票数 3
我正在尝试更新一个大的mysql表(有接近500百万行)的每一行的几个字段。表没有任何主键(或者有像UUID这样的字符串主键)。我没有足够的执行内存来一次读取和保存整个数据。谁可以让我知道我有什么选择来处理这样的表。
下面是模式
CREATE TABLE Persons ( Personid varchar(255) NOT NULL, LastName varchar(255) NOT NULL, FirstName varchar(255) DEFAULT NULL, Email varchar(255) DEFAULT NULL, Age int(11) DEFAULT NULL) )
浏览 1提问于2021-04-17得票数 0
我正在像这样应用SQL查询：
s"SELECT * FROM my_table_joined WHERE (timestamp > '2022-01-23' and writetime is not null and acceptTimestamp is not null)"
我收到了这样的错误信息。
warning: there was one deprecation warning (since 2.0.0); for details, enable `:setting -deprecation' or `:replay -deprecati
浏览 3提问于2022-01-23得票数 0
回答已采纳
2回答
我知道我们可以通过以下命令从SparkSQL查询或创建Mysql表。
val data = sqlContext.read.jdbc(urlstring, tablename, properties)
data.write.format("com.databricks.spark.csv").save(result_location)
val dataframe = sqlContext.read.json("users.json")
dataframe.write.jdbc(urlstring, table, properties)
像这样，有什么方法可以删
浏览 0提问于2016-02-24得票数 3
1回答
在SparkR 1.6.0中，我可以使用以下代码从JDBC源中读取数据，
jdbc_url <- "jdbc:mysql://localhost:3306/dashboard?user=<username>&password=<password>"
df <- sqlContext %>%
loadDF(source = "jdbc",
url = jdbc_url,
driver = "com.mysql.jdbc.Driver
浏览 3提问于2017-08-16得票数 2
回答已采纳
我用列数据类型- [Comments] [varchar](2000) NULL创建了Synapse表
当我使用Databricks火花代码写入Synapse时，我看到Synapse表列的数据类型更改为[Comments] [nvarchar](256) NULL。在Databricks dataframe上，数据类型为string。
数据库星火代码-
df.write \
.format("com.databricks.spark.sqldw") \
.option("url", "jdbc:sqlserver://<&
浏览 13提问于2022-07-18得票数 0
回答已采纳
我正在使用Spark Dataframe中的numPartitions，lowerBound，upperBound从oracle.but中获取大表，它给我的数据只相当于一个分区。假设分区数为10，记录总数为100，那么我在Hive中只能摄取10条记录(记录总数/numPartitions)。下面是我的代码片段
val hiveContext = SparkApp.getHiveContext("AppName")
val jdbcUsername = "MYUSERNAME"
val jdbcPassword = "MYPASSWORD"
浏览 0提问于2018-07-02得票数 1
已解决: prop.setProperty("driver"，"oracle.jdbc.driver.OracleDriver")必须将此行添加到连接属性中。
我正试着在当地找份火种工作。我用maven创建了一个带有依赖项的jar。
这是我的pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org
浏览 122提问于2016-07-08得票数 4
回答已采纳
我在Databricks中工作，试图使用sparklyr函数spark_write_jdbc向Server表写入数据。服务器名称/驱动程序等是正确的和工作的，因为我在前面的代码中成功地使用了sparklyr::spark_read_jdbc()。
根据文档()，spark_write_jdbc应该接受Spark。
我使用SparkR::createDataFrame()将我正在使用的dataframe转换为Spark dataframe。
以下是相关代码：
events_long_test <- SparkR::createDataFrame(events_long, schema =
浏览 2提问于2020-09-11得票数 0
我尝试将PCA应用于我的数据，然后将RandomForest应用于转换后的数据。然而，PCA.transform(数据)给了我一个DataFrame，但我需要一个mllib LabeledPoints来供给我的RandomForest。我该怎么做呢？我的代码：
import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.mllib.tree.RandomForest
import org.a
浏览 0提问于2016-03-13得票数 14
回答已采纳
我有一个包含以下数据的Spark数据帧(我使用spark-csv加载数据)：
key,value
1,10
2,12
3,0
1,20
有没有类似于spark RDD reduceByKey的东西，它可以返回Spark DataFrame为：(基本上，对相同的键值进行求和)
key,value
1,30
2,12
3,0
(我可以将数据转换为RDD并执行reduceByKey操作，但是有没有更好的Spark DataFrame API方法呢？)
浏览 1提问于2015-12-13得票数 11
回答已采纳
1回答
我有一个dataframe，它包含来自Hive的属性表和主表。我想删除列，然后应用掩蔽逻辑(SHA2)。
将postgre中的属性配置读取为Spark/scala作业中的Dataframe。
val propertydf = loading the property dataframe from postgre db
主蜂箱表
输出应该是
任何人，请帮助我用Spark/Scala编写代码。我无法转换ListString并将其从dataframe传递给函数。
浏览 0提问于2021-03-26得票数 0
回答已采纳
1回答
我有一张甲骨文表格，里面有一些俄语记录。
当我使用Spark JDBC读取这个表时，我收到的dataframe的值不正确。
你知道为什么会发生这种情况，以及如何修复它吗？
// executes given query using jdbc
def executeQuery(spark: SparkSession, configuration: Map[String, String], sql_statement: String): DataFrame
= spark.read.format("jdbc")
.option("driver", &
浏览 1提问于2019-11-26得票数 2
1回答
使用jdbc触发执行
、、、、
在Spark dataframe中，假设我从oracle获取数据，如下所示。 查询会完全在oracle中发生吗？假设查询很大。那么，这对oracle来说是一种开销吗？更好的方法是读取单独的dataframe中的每个过滤表数据，并使用Spark SQL或dataframe连接它，以便在spark中进行完整的连接？你能帮个忙吗？ df = sqlContext.read.format('jdbc').options(
url="jdbc:mysql://foo.com:1111",
dbtable="(SELECT * FROM abc,bcd....
浏览 8提问于2020-09-17得票数 0
我以DataFrame array<string>的身份处理事务
transactions: org.apache.spark.sql.DataFrame = [collect_set(b): array<string>]
我想将其更改为RDD[Array[string]]，但是当我将其更改为RDD时，它被更改为org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]
val sam: RDD[Array[String]] = transactions.rdd
<console>:42: error: type m
浏览 14提问于2017-01-11得票数 1
我使用的是spark 1.6，当我运行以下代码时遇到了上面的问题：
// Imports
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SaveMode
import scala.concurrent.ExecutionContext.Implicits.global
import java.util.Properties
import scala.concurrent.Future
// Se
浏览 2提问于2016-01-11得票数 16
我正在尝试运行这样一个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
回答已采纳
我一直在尝试使用sqlContext.read.format("jdbc").options(driver="org.apache.hive.jdbc.HiveDriver")将蜂窝表放入Spark，但没有任何成功。我做了以下的研究和阅读：
我使用了最新的Hortonworks Sandbox 2.6，并向那里的社区提出了同样的问题：
我想通过pyspark做的事情非常简单
df = sqlContext.read.format("jdbc").options(driver="org.apache.hive.jdbc.HiveDr
浏览 0提问于2018-01-04得票数 22
使用HDP 3.1中的Spark 2.4和Hive 3.1.0，我试图使用spark sql从hive读取托管表，但它返回一个空的dataframe，而它可以轻松地读取外部表。如何通过spark sql从hive读取托管表？ 注意:当从hive客户端读取时，hive maanged表不是空的。 1-我尝试通过ORC和拼花来格式化表格，但两者都失败了。 2-我无法使用HWC读取它。 3-我在使用JDBC时无法读取它。 os.environ["HADOOP_USER_NAME"] = 'hdfs'
spark = SparkSession\
.build
浏览 26提问于2019-09-25得票数 1
让我的dataframe如below.Here所示，我必须从我的dataframe的所有列名中删除字符串"_value“的最后一次出现。
import spark.implicits._
import org.apache.spark.sql.functions._
val simpledata = Seq(("file1","name1","101"),
("file1","name1","101"),
("file1","name1","101&#
浏览 4提问于2022-06-18得票数 1
我在MSSQL中有一个名为dbo.1table的表，我需要将它转换为dataframe，然后将它保存为avro文件，但我甚至不能将它加载为dataframe。
我用字段名为a的表测试了我的代码，它起作用了，我尝试转换表名"toString()“，但是到目前为止还没有什么效果。我希望有一个dataframe，然后将其保存为avro文件。相反，我有以下错误：
val DFDimAccountOperator = spark.read.format("jdbc")
.option("driver", "com.microsoft.sqlserve
浏览 0提问于2019-08-19得票数 0
回答已采纳
我可以使用简单的JDBC连接程序连接到Informix数据库，但是当我尝试使用Spark Dataframes加载表时，我遇到了一个异常。我们需要为Informix spark连接使用特定的连接器吗？
下面是异常的堆栈跟踪：
java.sql.SQLException: System or internal error java.lang.NumberFormatException: For input string: "table_name"
at com.informix.util.IfxErrMsg.getSQLException(IfxErrMsg.java:482)
浏览 0提问于2016-01-26得票数 3
1回答
[我们正在测试一个驱动程序，该驱动程序在优化后可能会提供出色的并行性。诀窍是，它不会在Spark分区内并行化(在访问DB2时)，所以要求我们告诉它我们想要多少个并行线程，并为每个线程抛出一个查询。虽然我希望在包含DataFrame对象数组的循环中做到这一点，但我不知道如何使用DataFrame对象数组编写scala。我做了一个暴力测试：
val DF1 = sqlContext.read.format("jdbc"). ...yada yada
val DF2 = sqlContext.read.format("jdbc"). ...yada y
浏览 0提问于2016-03-04得票数 1
2回答
我正在运行一个要加载到Oracle的spark作业。但是我得到了下面的错误。
java.sql.SQLException: ORA-00604: error occurred at recursive SQL level 1
ORA-01882:未找到时区区域
at oracle.jdbc.driver.T4CTTIoer.processError(T4CTTIoer.java:450)
at oracle.jdbc.driver.T4CTTIoer.processError(T4CTTIoer.java:392)
at oracle.jdbc.driver.T4CTTIoer.process
浏览 0提问于2018-05-23得票数 2
1回答
如何删除“。”从星火DataFrame的列名？
重命名具有“”的列名的DataFrame.select(F.col().alias())方法。在他们身上抛出一个错误。
下面的代码是可复制的。
# import Spark libraries, configuration, Contexts, and types.
import pyspark
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
#############
# Start Spark.
spark = SparkSession.builde
浏览 12提问于2017-11-15得票数 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
回答已采纳
在从postgres读取表格后，我在spark中创建了一个dataframe，如下所示。 val url = "jdbc:postgresql://localhost:5432/testdb"
val connectionProperties = new Properties()
connectionProperties.setProperty("Driver", "org.postgresql.Driver")
connectionProperties.setProperty("Username", "testD
浏览 19提问于2019-09-16得票数 0
回答已采纳
使用snowflake spark连接器从snowflake复制表时，默认行为是将结构化数据映射到spark字符串：https://docs.snowflake.net/manuals/user-guide/spark-connector-use.html#from-snowflake-to-spark-sql 例如，给定snowflake中的一个表： create table schema.table as
select
array_construct('1','a') as array_col,
object_construct(
浏览 26提问于2019-05-30得票数 1
Spark JDBC谓词下推禁用不起作用。
sparkVersion = "2.4.2“
已尝试：
properties.setProperty(JDBCOptions.JDBC_PUSHDOWN_PREDICATE, "false")
val dataFrame = sqlContext.read.jdbc(endpoint.connectionString, dbTableName, strictPredicates.get, properties)
dataFrame.createOrReplaceTempView(tableName)
谓
浏览 22提问于2019-07-18得票数 0
我正在尝试使用以下初始化脚本为Databricks设置外部配置单元元存储区：
dbutils.fs.put(
"/databricks/scripts/external-metastore.sh",
"""#!/bin/sh
|# Loads environment variables to determine the correct JDBC driver to use.
|source /etc/environment
|# Quoting the label (i.e. EOF) with single quotes to disabl
浏览 2提问于2021-02-18得票数 0
我想要使用DataFrame从关系数据库中读取sparkSession.read.jdbc(...)的单元测试代码。但是，我没有找到一种方法来模拟DataFrameReader来返回虚拟DataFrame进行测试。
代码示例：
object ConfigurationLoader {
def readTable(tableName: String)(implicit spark: SparkSession): DataFrame = {
spark.read
.format("jdbc")
.option("url", s
浏览 1提问于2019-04-03得票数 5
回答已采纳
1回答
迭代的列并更新指定的值
、、、
为了迭代从Hive表创建的Spark列并更新所有所需的列值，我尝试了以下代码。
import org.apache.spark.sql.{DataFrame}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions.udf
val a: DataFrame = spark.sql(s"select * from default.table_a")
val column_names: Array[String] = a.columns
val required
浏览 0提问于2018-05-06得票数 0
回答已采纳
我正在尝试从我的Spark作业中写出一些元数据到postgres数据库。继续输入以下错误：
2018-03-13 15:23:40,599 root ERROR An error occurred while calling o1315.jdbc.
: scala.MatchError: null
at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:62)
at org.apache.spark.sql.
浏览 1提问于2018-03-14得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
