在sql查询中调用DataFrames
在SQL查询中调用DataFrames是指在使用SQL语句进行数据查询时，可以将查询结果以DataFrame的形式返回。DataFrame是一种二维表格数据结构，类似于关系型数据库中的表，它可以方便地进行数据处理和分析。
在云计算领域中，调用DataFrames可以带来以下优势：
- 数据处理灵活性：DataFrames提供了丰富的数据处理函数和操作，可以进行数据清洗、转换、过滤、聚合等操作，方便进行复杂的数据处理任务。
- 数据分析能力：DataFrames集成了各种数据分析工具和库，如Pandas、NumPy等，可以进行统计分析、机器学习、数据挖掘等任务，帮助用户从数据中发现有价值的信息。
- 数据可视化：DataFrames可以与各种数据可视化工具结合，如Matplotlib、Seaborn等，可以方便地将数据可视化展示，帮助用户更直观地理解数据。
- 跨平台支持：DataFrames可以在不同的编程语言和平台上使用，如Python、R、Scala等，可以方便地与其他工具和系统集成，实现数据的无缝交互和共享。
在实际应用中，调用DataFrames可以应用于以下场景：
- 数据分析和报表：通过调用DataFrames，可以方便地进行数据分析和生成报表，帮助企业了解业务情况、优化决策。
- 数据清洗和预处理：通过调用DataFrames的数据处理函数，可以对原始数据进行清洗、去重、填充缺失值等操作，提高数据质量。
- 数据挖掘和机器学习：通过调用DataFrames和相关的机器学习库，可以进行数据挖掘、模型训练和预测，帮助企业发现潜在的商机和优化业务流程。
- 实时数据分析：通过调用DataFrames和流式处理框架，如Apache Kafka、Apache Flink等，可以实现实时数据分析和监控，帮助企业及时发现和解决问题。
腾讯云提供了一系列与DataFrames相关的产品和服务，如云数据库TDSQL、云数据仓库CDW、云数据湖CDL等，这些产品可以帮助用户在云上快速搭建和管理数据分析环境，实现高效的数据处理和分析。具体产品介绍和链接地址可以参考腾讯云官方网站：https://cloud.tencent.com/product
相关·内容
4回答
我是Spark世界的新手。我目前正在迁移我的应用程序的摄取代码，其中包括在HDFS中摄取数据，在HDFS中使用原始数据和应用层，并执行CDC(变更数据捕获)，这是目前在Hive查询中编写的，并通过Oozie执行。这需要迁移到Spark应用程序(当前版本1.6)中。代码的另一部分稍后将迁移。
在spark中，我可以直接从Hive中的表创建数据格式，只需按原样执行查询(如sqlContext.sql("my hive hql") )。另一种方法是使用dataframe并以这种方式重写hql。
这两种方法有什么不同？
使用Dataframe有什么性能提高吗？
有些人认为，在直接使用&
浏览 1提问于2017-08-01得票数 43
回答已采纳
我正试着理解火花hiveContext。当我们使用hiveContext编写查询时
sqlContext=new HiveContext(sc)
sqlContext.sql("select * from TableA inner join TableB on ( a=b) ")
它是使用火花引擎还是蜂巢引擎？我相信上面的查询是用火花引擎执行的。但是如果是这样的话，我们为什么需要数据文件呢？
我们可以盲目地复制sqlContext.sql("")中的所有hive查询并运行，而无需使用数据格式。
对于DataFrames，我的意思是像这样的TableA.join(
浏览 1提问于2017-09-14得票数 2
回答已采纳
我在SAS中有一个查询，其中我使用一个宏变量来用一个不同的变量重复对Teradata的查询。我们有5个数据库，每个状态一个，其中我运行相同的查询，但使用变量更新状态，然后修补所有数据集。我在寻求关于如何在python中这样做的帮助。
循环遍历{state1、state2、state3、state4、state5}，并将每个查询保存为{stateX}_df，然后合并所有查询
import teradata as td
import pandas as pd
from teradata import tdodbc
udaExec = td.UdaExec(appConfigFile="u
浏览 15提问于2020-02-19得票数 0
回答已采纳
2回答
我正在学习Scala中的Spark，它来自于大量的Python滥用，而且我得到了一个java.lang.NullPointerException，因为我是用python的方式来做事情的。
我说了每个4x2形状的数据，第一列总是索引0,1,2,3，第二列是一些二进制特性。最终的目标是拥有一个4x4的数据格式，并加入所有的独立数据。在python中，我首先定义一些主df，然后对中间的主df进行循环，在每个循环中将得到的联合dataframe分配给主dataframe变量名(丑陋)：
dataframes = [temp1, temp2, temp3]
df = pd.DataFrame(index
浏览 3提问于2018-10-24得票数 0
回答已采纳
Spark-shell：它基本上打开了scala>提示符。查询需要以下面的方式编写
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)
// Queries are expressed in HiveQL
sqlContext.sql("FROM src SELECT key, value").collect().foreach(println)
spark-sql：它似乎与蜂巢转移直接相连，我们可以用类似的方式编写查询。并查询蜂箱中的现有数据。
我想知道这两者的区别..。在spark中处理任何查询
浏览 7提问于2017-05-01得票数 6
回答已采纳
我正在修改Julia的代码，这些代码不是我自己写的。我在Julia做的工作很少。我连接了一个字符串，以创建一个从数据库返回单个整数值的SQL查询。我想在下一个sql字符串中使用返回值。但是，从初始SQL Select语句返回的值是1x1数据帧。我的假设是，我需要将返回值赋给一个整数变量，以便在后续的SQL字符串中使用。 import Dates
using ODBC
using Dataframes
dt = Dates.today()
db = ODBC.DSN("mydsn","myuser","mypasswd")
sqlstr = s
浏览 21提问于2021-11-13得票数 3
回答已采纳
1回答
我用RStudio完成了我的经济学论文，但是由于在这个过程中占用了大量内存，我的脚本非常慢。
My Case
我有一个庞大的数据集(股票价格，每天频率为10年，~700股，即3500美元，700美元)，我选择每只股票作为向量，将其分解为小波和CF滤波器(2个数据集，$28000\times700美元)，并应用benford定律(两个数据集，$9\times700美元)。
问题
RStudio将我的数据集存储在内存中，它们仅通过触摸就消耗了很大一部分数据。
问题
我开始学习基本SQL命令，发现我可以从某个表中调用特定的列。如果我一个一个地调用我的股票作为向量，而不是直接从RStudio中选择它们
浏览 0提问于2018-04-11得票数 0
回答已采纳
1回答
我必须使用Scala/Spark方法和SQL查询来操作RDD。
是否可以通过查询直接使用RDD 操作？
建议的方法(schemaRDD或DataFrame)需要额外的内存泄漏。
经过这样的转换，我记忆中有两个相同的巨大物体。
浏览 2提问于2016-02-08得票数 1
1回答
我是OOP新手，我正在尝试创建一个类和函数，它接收名称并为我创建SQL调用，比如：df_1 and df_2。
但是，我希望我的类为我管理它，所以args将只是重复名称，比如：('table1', 'table2', 'n'..)..
那我怎么才能用最好的方式做这件事呢？当我学习的时候，我很感激解释
import pandas as pd
from def_funcs import sup # I got my data base engine from this func
from typing import Tuple, List, Dic
浏览 4提问于2022-08-31得票数 1
回答已采纳
我使用在scala中返回true/false的一些业务逻辑，在中添加了一列。实现是使用UDF完成的，UDF有10个以上的参数，因此在使用UDF之前我们需要先注册UDF。已完成以下工作
spark.udf.register("new_col", new_col)
// writing the UDF
val new_col(String, String, ..., Timestamp) => Boolean = (col1: String, col2: String, ..., col12: Timestamp) => {
if ( ... ) true
浏览 0提问于2019-04-20得票数 0
我假设spark.sql(查询)是在使用spark.sql时使用的，而spark.read.format("jdbc").option("query“、"")则是在使用oracle语法时使用的。我这样假设是对的吗？
浏览 3提问于2021-08-16得票数 3
回答已采纳
1回答
我是斯派克和斯卡拉的新手。我正在尝试从SQL-server中的一个过程中获取内容，以便在Spark中使用它。为此，我将通过Scala (Eclipse)中的JDBCRDD导入数据，并从过程中生成一个RDD。
在创建RDD之后，我将它注册为一个临时表，然后使用sqlContext.sql(“选择查询以选择特定列”)。但是，当我在select查询中输入列名时，它会抛出一个错误，因为我在RDD和临时表中都没有列名。
请在下面找到我的代码：
val driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
val url = XXX
浏览 4提问于2015-04-11得票数 0
回答已采纳
在databricks中，我有一个笔记本，其中只有SQL查询，我想知道是否更好(谈论性能)将它们全部转换为pyspark，或者是否相同。
换句话说，我想知道databricks是否使用spark来执行查询。
我发现了问题(看起来与我的问题很相似)，但答案不是我想知道的。
浏览 10提问于2022-04-08得票数 0
我正在使用Python连接到SQL数据库。我设法关闭了数据库连接，但希望在一个函数中创建一个for循环，该函数读取字符串列表，循环它并创建多个数据帧，但无法考虑正确的逻辑。
def database_create(SQLscript,D_name):
global data
cur = con.cursor()
for i in SQLscript:
cur.execute(SQLscript)
D_name = pd.DataFrame(cur.fetchall(), columns = [des
浏览 1提问于2022-08-04得票数 0
1回答
在使用Python Teradata (tdodbc)查询时，我遇到了使用不同变量遍历同一查询并合并结果的问题。我在另一个中得到了很好的指导，并最终来到了这里。我现在的问题是，数据帧最终只得到循环中最后一个变量"state5“的查询结果。不幸的是，我们在自己的数据库中有5个州，每个州都有相同的模式。我可以运行相同的查询，但希望循环变量，这样我就可以运行所有5个状态并返回一个附加的查询。这很容易使用SAS宏变量和修补，但需要将数据带到python中进行EDA和数据科学。
from teradata import tdodbc
udaExec = td.UdaExec(appConfigF
浏览 39提问于2020-02-21得票数 1
回答已采纳
1回答
我经常将不同数据库中的数据提取到熊猫数据中，进行一些处理，然后最终执行一个我非常喜欢用SQL编写的处理步骤(通常涉及跨DFs的联接)。我想我会写一个快速的函数
dataframeswrite 接受一个任意的sql查询，并将相关的设置为tmp sqlite3 db，在sqlite3中执行sqlite3查询，并将结果作为数据格式返回。
我想出的函数如下所示：
def SQL_on_DFs(query:str, dfs=[]) -> pd.DataFrame:
"""convenience function to write sql on arbitrary D
浏览 7提问于2022-02-04得票数 2
回答已采纳
我一直在玩Apache Spark，首先我学习了PostgreSQL，我有几个问题需要在Spark上运行。我设法在Spark SQL中将它们作为SQL字符串运行，但现在我必须执行RDD操作才能获得相同的结果。我将数据从csv加载到地图。现在，我必须选择这些映射中的特定列，但我不知道如何连接它们(多个映射/csv文件)。我的第二个问题是如何最好地执行RDD操作，以便从postgresql查询中获得相同的结果？ 我试着阅读了RDD操作，其中包括转换和连接，但它不允许我连接映射。 其中一个查询： SELECT Tournaments.TYear,Countries.Name,Max(Matches
浏览 7提问于2019-06-09得票数 0
1回答
我正在将MongoDB连接到Spark，并希望使用查询加载数据。
df = sqlContext.read.format("com.mongodb.spark.sql").options(collection='test', query = {'name' :'jack'}).load()
df.show()
但它还给了我全部藏品。如何在星火中复制与此查询db.test.find({'name':'jack'})相同的内容？
浏览 2提问于2017-04-11得票数 2
回答已采纳
1回答
我有一个小的log数据帧，其中包含关于在给定笔记本中执行的ETL的元数据，该笔记本是在Azure DataFactory中管理的更大的ETL管道的一部分。 不幸的是，Databricks似乎不能调用存储过程，所以我手动将包含正确数据的行追加到我的日志表中。 但是，在给定一组条件的情况下，我找不到正确的sytnax来更新表： 我用来追加一行的语句如下： spark_log.write.jdbc(sql_url, 'internal.Job',mode='append') 然而，这工作得很顺利，因为我的数据工厂正在调用一个存储过程， 我需要在查询中工作，比如 qu
浏览 29提问于2020-01-06得票数 2
回答已采纳
2回答
我有一个与DB表值相比较的程序，我已经在PyQt5中创建了一个GUI。我已经创建了两个线程，一个用于查询每个表，然后程序必须等待两个线程完成。我下面的代码
from PySide2 import QtWidgets
from PySide2 import QtGui
from PySide2 import QtCore
from Main_interface import Ui_mainWindow
import pandas as pd
class mainWindow(QtWidgets.QMainWindow, Ui_mainWindow):
sqlClicked1 = QtC
浏览 0提问于2019-08-21得票数 0
回答已采纳
1回答
如何在熊猫数据中插入query_result输出？查询结果首先获得500个记录，并使用QueryMore提取其余记录。我无法将df中的所有内容连同列名一起推送。
#!/usr/bin/env python3
import beatbox
import sqlalchemy
engine_str = 'mysql+mysqlconnector://db-user:db-pass@localhost/db-name'
engine = sqlalchemy.create_engine(engine_str, echo=False, encoding='utf-8'
浏览 4提问于2016-11-14得票数 0
回答已采纳
1回答
我正在尝试使用pyspark运行sql查询。我使用以下代码：
from pyspark import SparkConf, SparkContext
from pyspark.sql import SQLContext, Row
import os
spark_config = SparkConf().setMaster("local").setAppName("Project_SQL")
sc = SparkContext(conf = spark_config)
sqlctx = SQLContext(sc)
df = sqlctx.read.forma
浏览 2提问于2017-09-04得票数 1
回答已采纳
我们使用python spark驱动程序V3.0.0。当尝试使用dataframe加载数据时，where子句无效。然而，CQL本身确实在Datastax DevCenter中工作。代码如下所示
dataf = sqlc.read.format("org.apache.spark.sql.cassandra")\
.options(table="tran_history", keyspace="test")\
.load()\
.where("usr_id='abc' log_ts >= maxtimeuuid(
浏览 0提问于2016-03-16得票数 1
我正在尝试从MS SQL表中读取数据，其中一列包含超过1024个字符的varchar。当我将查询读入数据帧时，字符串被截断为1023个字符。请参见下面的代码和输出。有人知道如何将超过1023个字符的字符串放入数据帧吗？我到处寻找，在DataFrame文档中，但我还没有找到答案。
在下面的代码中，表test包含一个列's‘，其中一行有一个长度为1100的字符串。列'len_s‘包含字符串's’的长度(在SQL server中计算)。
import pypyodbc
from pandas import Series, DataFrame
import pandas as
浏览 27提问于2017-01-22得票数 1
回答已采纳
我有一个要合并回SQL表的数据帧-不是熊猫意义上的合并，这将是一个连接，而是一个SQL合并操作，根据数据帧和表之间的比较更新/插入表中的记录。
我可以看到一些替代方法，比如将数据帧写入新表并在SQL中进行合并，或者删除基于主键的任何现有记录并使用to_sql将整个数据帧附加到表中-但是在python中是否有内置的函数可以直接执行这种合并？
浏览 20提问于2020-05-17得票数 1
1回答
我有两个dataframes (Scala )A和B。当A("id") == B("a_id")时，我想将A("value")更新为B("value")。由于DataFrames必须重新创建，所以我假设我必须执行一些联接和withColumn调用，但我不知道如何做到这一点。在SQL中，这将是对一个自然联接的简单更新调用，但由于某种原因，这在Spark中似乎很困难吗？
浏览 0提问于2017-03-17得票数 1
回答已采纳
5回答
我是spark.Is的新手，不管怎么说，我可以在spark-SQL中操作更新命令。我已经创建了一个hive表，我已经在其中操作了一些SQL语法查询，如insert、select、delete，但无法运行update命令。我已经在spark文档和网站上搜索过了。但是我没有在他们最近的Spark SQL列表中找到UPDATE Syntax。请帮帮我，我用的是spark 2.0
卡里安
浏览 27提问于2016-08-09得票数 2
1回答
我对Spark很陌生，但我知道hive查询执行框架。我想了解spark如何执行sql查询(技术描述)？
如果我在命令下开火
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)
sqlContext.sql("select count(distinct(id)) from test.emp").collect
在蜂巢中，它将被转换成地图-减少作业，但是它是如何在星火中执行的呢？
蜂巢亚稳态将如何进入画面？
提前谢谢。
浏览 1提问于2016-03-23得票数 2
1回答
我正在为数据准备工作的火花-sql。
我面临的问题是在获得sql查询结果之后。我应该如何根据If-然后-else条件来更新行。
我在做什么
val table_join = sqlContext.sql(""" SELECT a.*,b.col as someCol
from table1 a LEFT JOIN table2 b
on a.ID=b.ID """)
table_join.registerTempTable("Table_join")
现在，当我有最后一个连接表，它是df格式的。如何更新行？
//Fi
浏览 2提问于2015-08-12得票数 1
回答已采纳
2回答
迭代select查询
、、、
我想创建一个数据集，它是许多select查询的堆叠数据集，在我的where子句中更改了一个参数。我可以使用python通过调用select查询n次并相应地更改where原因来做到这一点。这是我可以在SQL中一次性完成的事情，而不必进行n个单独的select查询吗？ 例如 dfs = []
for name in names:
dfs.append(spark.select("select * from personal_data where Name = 'f{name}'"))
# Some logic to take the list of dat
浏览 31提问于2019-12-10得票数 0
2回答
我们的需求是对Phoenix (HBase)时间序列表执行一些分析操作。我们在postgresql中有一个表，其中包含uniqueIds。
现在，我们正在从postgresql表中获取所有uniqueIds，并在Phoenix表中查询相应的uniqueIds，并在这里应用分析functions.But。所有uniqueIds都是以顺序方式处理的。我们需要让它并行运行。我们正在使用scala和spark来实现此功能。
下面是示例代码，
val optionsMap = Map("driver" -> config.jdbcDriver, "url" ->
浏览 2提问于2017-05-22得票数 0
1回答
我正在读取csv并使用以下过程将其插入到云sql中：
df = pd.read_csv(csv_file, sep=';', encoding='utf-8', keep_default_na=False)
## from field “world” remove C. , County , Cnty
df['world'] = df['world'].str.rstrip('C.CountyCnty')
## connects to mysql database and adds t
浏览 1提问于2018-01-25得票数 0
回答已采纳
1回答
我想从2020年到现在运行对日期的SQL查询。我可以在这个循环中列出这个列表：
rep_dates = list(pd.date_range(start='2020-07-31',
end='2022-09-30',
freq='1M').strftime('%Y-%m-%d'))
df_list = [0 for i in range(len(rep_dates))]
for i in range(len(rep_dates)):#, obs_date in enumer
浏览 3提问于2022-10-04得票数 -1
2回答
我使用Spark structured streaming处理从Kafka读取的记录。以下是我想要达到的目标：
(a)每条记录都是Tuple2类型的(Timestamp, DeviceId)。
(b)我已经创建了一个静态的Dataset[DeviceId]，它包含了期望在Kafka流中看到的所有有效设备in (类型为DeviceId)的集合。
(c)我需要编写一个Spark structured streaming查询
(i) Groups records by their timestamp into 5-minute windows
(ii) For each window, get
浏览 2提问于2017-10-02得票数 15
因此，我有以下场景：
我有一个服务，可以对数据帧中的参数进行一些计算。例如，我正在执行describe()操作。我通过和http-post (ArrayString+schema)获取参数，并通过read.json函数在sql上下文中读取它们。
我可以在一个有10.000个参数的大数据帧中得到它，也可以在只有一个参数的10.000个小数据帧中得到它。每行大约有12.000行，带有时间戳。最后，我需要收集数据帧以将其发送到不同的服务进行进一步计算。由于输入的创建方式，对它进行参数设置会更容易。
但我发现，在许多小的数据帧上进行收集/转换为json比在一个巨大的数据帧上进行收集/转换要昂贵得多。对
浏览 0提问于2016-03-15得票数 2
1回答
给定一组文件--每个文件都被读取到一个不同的dataframe中-- pandasql查询如何引用它们？
在下面的片段中，我们有一个list of dataframes：但是同样的问题也适用于dict
import pandas as pd
from pandasql import sqldf
# Read in a set of 10 files each containing columns `id` and `estimate`
dfs = [pd.read_csv('file%d.csv' %d) for d in range(1,10+1)]
sql_res = s
浏览 1提问于2018-11-17得票数 1
回答已采纳
2回答
我使用spark sql v2.3.1将数据写入cassandra v3.11.1。要将数据帧写入cassandra表，我们可以使用
df.write.format("org.apache.spark.sql.cassandra")
.options(Map( "table" -> tname, "keyspace" -> keyspacename ))
.mode(SaveMode.Append)
.save();
但是使用相同的版本，我如何使用saveToCassandra()方法，我需要添加任何jars，我已经添加了火花
浏览 1提问于2018-10-17得票数 2
回答已采纳
我有一个Seq of Spark (即Seq[org.apache.spark.sql.DataFrame])，它可以包含一个或多个元素。有一个列的列表对每个数据文件都是通用的，每个dataframe也有一些额外的列。我想要做的是使用连接条件中的那些公共列将所有这些数据文件连接在一起(记住，数据格式的数量是未知的)
我怎样才能把所有这些数据文件连接起来？我想我可以对它们进行foreach，但这似乎不太优雅。谁能想出一种更实用的方法来做这件事呢？编辑:递归函数会比预测更好，我现在正在做这个工作，完成后会在这里发布。
下面是一些代码，它创建了n个dataframes列表(在本例中是n=3)，每个列
浏览 0提问于2018-05-15得票数 4
回答已采纳
我尝试将在数组中获取的列名传递给spark sql，但不确定如何将其作为变量传递。我已经创建了一个dataframe并获取了变量x中的所有列，现在我创建了一个foreach循环来创建一个查询字符串，以便在下一步中进行查询。我们该如何解决这个问题？最终结果应该是"dftable“中各个列的总和。请帮帮忙。
df.createOrReplaceTempView(“dftable”)
// collect an array of columns
val x= df.schema.fieldNames
// create query string for the
浏览 3提问于2017-07-19得票数 2
回答已采纳
1回答
我们正试图在EMR中运行火星上的ETL。在S3中，大约有2000万个事件被压缩为json行。它们总共大约有30个文件。我用的是火花放电。
这是密码，
def value_to_list(columns):
def value_map(values):
data = []
for val in values:
d = val.asDict()
data.append([d[column] for column in columns])
return data
return value
浏览 2提问于2016-11-15得票数 0
回答已采纳
我用的是火花1.3.1。在PySpark中，我从RDD中创建了一个DataFrame并注册了模式，如下所示：
dataLen=sqlCtx.createDataFrame(myrdd, ["id", "size"])
dataLen.registerTempTable("tbl")
此时，一切都很好，我可以从" tbl“中进行"select”查询，例如“从tbl选择大小，其中id='abc'”。
然后，在Python函数中，我定义了如下内容：
def getsize(id):
total=sqlCtx.s
浏览 4提问于2015-06-04得票数 1
我想知道spark中的数据帧缓存是否是线程安全的。在我们的一个用例中，我从一个hive-table创建一个dataframe，然后通过不同的线程在同一个dataframe上运行多个SQL。由于我们的存储和计算是解耦的，而且由于某些原因读取非常慢，我在考虑将数据帧缓存到内存中，并将缓存的数据帧用于所有查询。数据帧缓存是线程安全的吗？这样做还有其他的陷阱吗？我的计算集群中有足够的内存(磁盘和内存)来缓存表，我将在相同的数据帧上执行10+查询。
谢谢,
阿卡什
浏览 17提问于2020-03-10得票数 0
2回答
遍历数据表的行。
、、、
因为我对Scala有点陌生，所以我发现很难遍历Dataframe。我的dataframe包含两个列，一个是path，另一个是ingestiontime。例子-
现在，我想迭代这个数据，并使用Path和ingestiontime列中的数据来准备一个Hive查询并运行它，这样运行的查询看起来就像-
ALTER TABLE <hiveTableName> ADD PARTITON (ingestiontime=<Ingestiontime_From_the_DataFrame_ingestiontime_column>) LOCATION (<Path_Fro
浏览 2提问于2020-07-22得票数 0
回答已采纳
我试图将一堆数据文件导出到SQL表中，跳过有错误的表，但是得到了这个错误："TypeError：'DataFrame‘对象是可变的，因此它们不能被散列。“
dict = {df1:'tableName1',
df2:'tableName2',
df3:'tableName3'}
for df in dict:
try:
df.to_sql(dict[df],engine,index=False,method=functools.partial(pd_writer,quote_ident
浏览 5提问于2022-10-03得票数 -2
回答已采纳
我对下面这段代码有问题：
def skewTemperature(cloudantdata,spark):
return spark.sql("""SELECT (1/count(temperature)) * (sum(POW(temperature-%s,3))/pow(%s,3)) as skew from washing""" %(meanTemperature(cloudantdata,spark),sdTemperature(cloudantdata,spark))).first().skew
meanTemperature和
浏览 2提问于2017-09-06得票数 1
1回答
我需要加入两个星火数据，并将结果返回蜂巢。以下是数据格式：
Dataframe1: Cassandra表-分区和集群键：(ID，PART_NBR)
val df1 = spark.read.format("org.apache.spark.sql.cassandra")
.option("keyspace", "mykeyspace")
.option("table", "mytable")
.load
** Dataframe 2:从另一个来源获得的键的数据(即上表中的ID列的分区键)--
浏览 3提问于2020-08-11得票数 1
我有以下SparkSQL：
val resultDf = spark.sql("SELECT name, phone, country FROM users")
我想对下列收藏中存在的国家返回的记录进行过滤：
val countries = Seq("Italy", "France", "United States", "Poland", "Spain")
例如，我可以根据集合创建广播变量：
val countriesBroadcast = sc.broadcast(countries)
但这是否有
浏览 0提问于2018-11-09得票数 0
回答已采纳
2回答
我在csv的一栏中列出了一组表，我把这些表格作为数据输入到熊猫中，如下所示：
df = pd.read_csv('~/tablenames')
表名csv如下所示：
tablenames
table1
table2
创建到db的连接并遍历dataframe如下所示：
import pandas as pd
from sqlalchemy import create_engine
df = pd.read_csv('~/tablenames')
engine = create_engine('connection_string')
f
浏览 1提问于2019-05-13得票数 0
回答已采纳
1回答
我试图从一个数据读取数据，并在另一个数据中使用它。我怎么才能优雅地做这件事？
val query = s"select distinct p_id, lower(regexp_replace(p_id,'[^a-zA-Z0-9]+','_')) as p_id_formatted, lower(regexp_extract(f_id,'^([^\\.]+)\\.?',1)) as f_id_formatted, column_name from default.rc_pcoders"
val run_query = sql(que
浏览 0提问于2018-03-21得票数 1
回答已采纳
1回答
我试图使用在Cassandra的数据基础上进行分析，并找到了两种类型的实现。有人能对两者之间的区别和优势/劣势提出一些看法吗？我正在尝试查看用于查询大型数据集的哪一个。谢谢
选项1-使用Session SQL
sparkSession.read
.format("org.apache.spark.sql.cassandra")
.options(Map( "table" -> table, "keyspace" -> keyspace))
.load()
选项2-使用SCC
CassandraJa
浏览 3提问于2021-06-20得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
