我们有下面的查询，它在大查询环境中工作。AS FLOAT64) > 0 我需要使用python在spark环境中运行该查询。from pyspark.sql import SparkSessiondf.createOrReplaceTempView("people")
df2=spark.sql("""SELECT id,name, SAFE_CAST(value AS
浏览 105提问于2019-12-18得票数 0
回答已采纳
datasetName.tableName选择SQRT((D_o_latitude - T_s_lat)^2+(D_o_longitude - T_s_long)^2)/0.00001
所以请告诉我，我们能不能将字符串类型转换为在查询中浮动，比如转换数据类型，我不能更改数据类型。
浏览 3提问于2014-04-05得票数 43
我正在尝试使用AWS Lambda函数运行我的PySpark作业代码。下面是错误:无法导入模块'lambda_function'：没有名为'pyspark‘的模块from
浏览 0提问于2018-12-11得票数 0
2回答
如何在PySpark SQL中实现用户定义的聚合函数？pyspark version = 3.0.2作为一个最小的例子，我想用一个UDAF替换AVG聚合函数：sql将位于的位置：Out[2]:0 1 1.5在查询中，如何使用U
浏览 24提问于2021-03-09得票数 4
回答已采纳
我目前正在自学Spark programming，并试图用PySpark重新编写一个现有的Python应用程序。然而，我仍然对如何在PySpark中使用常规Python对象感到困惑。我了解Spark中的分布式数据结构，如RDD、DataFrame、Datasets、vector等。Spark有自己的转换操作和动作操作，如.map()、.reduceByKey()来操作这些对象。但是，如果我在PySpark中创建传统的Python数据对象，比如数组、列表
浏览 34提问于2017-03-01得票数 1
回答已采纳
如何在DataFrame中为流PySpark设置架构。from pyspark.sql import SparkSessionfrom pyspark.sql.functionsimport splitfrom pyspark.sql.types import *
spa
浏览 3提问于2016-12-29得票数 3
回答已采纳
1回答
我需要编写一个自定义的pySpark，我遇到了这个例子。在类似的行中，如线程的最后一部分所示，我提出了以下函数 StructField("keybloomfilter
浏览 1提问于2019-04-04得票数 1
PySpark的文档在示例中包括以下内容：from pyspark.sql.functions import *from datetime import date, timedelta, datetime
我不认识或理解最后两行的语法。我知道导入路径中的.和.. (“相对导入路径”)，但是这个语法对我来说是
浏览 5提问于2022-06-15得票数 0
回答已采纳
在Google BigQuery中，我有一个时间戳字段，它的年份为20195。这当然会导致错误，因为它超出了标准SQL时间戳范围。但是，即使使用SAFE_CAST，我也不能更新这些记录或删除此记录作为错误。例如，我试过：
UPDATE [table] SET DateField = SAFE_CAST('2019-01-01 00:00:00 UTC' AS TIMESTAMP)...问题是:如何在没有任何错误的情况下成功地删除或更新？
浏览 1提问于2019-03-02得票数 0
我正在尝试使用python从文本文件中删除重复的数字，但该操作仅适用于行。例如，我的文本文件是：66 9 23 import pyspark
from pyspark import SparkContext, SparkConf
浏览 2提问于2022-02-14得票数 0
回答已采纳
我知道我可以用 %pyspark
df = sqlContext.sql('select * from train_table') 我可以使用df.registerTempTable('xxx')使df在%sql中可访问。但是有时候我想用%sql来画图。移到df = sqlContext.sql(sql)中，重新计算。我想知道有什么方法可以在%pyspark中访问%sq
浏览 30提问于2019-02-15得票数 0
2回答
为此，我打算使用此函数： def sum_col(data, col):
return data.select(f.sum(col)).collect()[0][0] 我现在正在考虑做这样的事情
浏览 13提问于2020-07-01得票数 3
回答已采纳
我正在处理大量的数据，这些数据无法通过PySpark中的可用内存进行处理，这会导致内存不足。为此，我需要使用MEMORY_AND_DISK选项。我的问题是:如何在PySpark朱庇特笔记本中启用这个标志？我在找这样的东西： .master('local[*]'是否有类似的
浏览 3提问于2020-06-09得票数 0
回答已采纳
7回答
在pyscam1.6.2中，我可以通过以下方式导入col函数但是当我试图在中查找它时，我发现functions.py文件中没有col函数，python怎么能导入一个不存在的函数呢？
浏览 9提问于2016-10-20得票数 91
回答已采纳
df = df.withColumn("day", df.date_time.getField("day"))
org.apache.spark.sql.AnalysisException: GetField
浏览 0提问于2015-06-17得票数 18
回答已采纳
1回答
我正在尝试用spark做一个结构化的流媒体应用程序，主要思想是从kafka源读取，处理输入，写回另一个主题。我已经成功地让spark读写了kafka，但是我的问题出在处理部分。我尝试过foreach函数来捕获每一行并在写回kafka之前对其进行处理，但是它总是只做foreach部分，而不会写回kafka。但是，如果我从writestream中删除foreach部分，它将继续写入，但现在我丢失了处理。
如果有人能给我举一个例子，告诉我如何做到这一点，我将非常感激。
浏览 37提问于2019-07-17得票数 0
回答已采纳
5回答
- hours: struct (nullable = true)我希望创建一个新的数据框架(new_df)，以便'name'列中的值不包含任何空格我的代码是：from pyspark.sql import SQLContext
from pyspark.sql import HiveContextfrom pyspark
浏览 10提问于2016-02-21得票数 9
回答已采纳
df = sqlContext.sql("select d1.a, d1.b, d1.c as aaa, d2.d, d2.e, d2.f, d2.g, d2.h, d2.i, d2.j as lengthd2.j and length(upper(trim(d1.a))) = (d1.j+3)".format(dataBase, month_end)) File "/opt/cloudera
浏览 1提问于2018-05-25得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
