使用spark-sql或pyspark模式在列之间匹配的转换
可以通过DataFrame的函数和操作来实现。
首先，让我们简单介绍一下Spark和Pyspark。Spark是一个用于大规模数据处理的开源分布式计算框架，它提供了丰富的API，包括Spark SQL用于结构化数据处理。Pyspark是Spark的Python API，提供了与Spark的交互式编程能力和数据处理能力。
在Spark SQL中，可以使用DataFrame API和SQL语言来进行数据处理。DataFrame是一种由行和列组成的分布式数据集，类似于传统数据库表。在DataFrame中，可以使用函数和操作来进行列之间的匹配转换。
要在列之间进行匹配转换，可以使用以下函数和操作：
- selectExpr(): 这个函数可以使用SQL表达式选择列，并且可以通过AS关键字给列起别名。例如，可以使用selectExpr("col1", "col2 + 1 AS new_col")来选择col1列和将col2加1的结果作为new_col列。
- withColumn(): 这个函数可以添加新列或替换现有列，并且可以使用列之间的表达式进行转换。例如，可以使用withColumn("new_col", col("col1") + col("col2"))来添加一个名为new_col的新列，其值是col1和col2列之和。
- select(): 这个函数可以选择指定的列，并返回一个新的DataFrame。例如，可以使用select("col1", "col2")来选择col1和col2列。
- alias(): 这个函数可以为列添加别名。例如，可以使用alias("new_col")将列重命名为new_col。
- when(): 这个函数可以根据条件进行列值的转换。例如，可以使用when(col("col1") > 10, 1).otherwise(0)来将大于10的col1列的值转换为1，否则转换为0。
- cast(): 这个函数可以将列的数据类型进行转换。例如，可以使用cast("col1", "integer")将col1列的数据类型转换为整数类型。
在具体的应用场景中，根据数据的结构和需求，可以使用上述函数和操作来进行列之间的匹配转换。例如，可以根据特定的业务需求选择指定的列，并进行列值的计算、转换、重命名等操作。
对于腾讯云的相关产品和产品介绍链接，我不能提及具体的产品名称，但你可以访问腾讯云官方网站，了解他们提供的云计算服务和解决方案，以找到适合你的需求的产品。
希望以上信息对你有所帮助！
相关·内容
我正在尝试使用SPARK-SQL和pyspark包装器来实现。我遇到了这个错误
pyspark.sql.utils.AnalysisException: u"cannot resolve '```(qtr)?+.+```'
given input columns:
这是我的query..Basically试图排除列'qtr‘。
select `(qtr)?+.+` from project.table;
使用以下属性在hive/beeline中工作得很好
set hive.support.quoted.identifiers=none;
有什
浏览 1提问于2018-09-27得票数 3
我正在尝试使用列‘'$NUMBER'’中的值来替换各行的子字符串number。我试过了
from pyspark.sql.functions import udf
from pyspark.sql.Types import StringType
replace_udf = udf(
lambda long_text, number: long_text.replace("$NUMBER", number),
StringType()
)
df = df.withColumn('long_text',replace_udf(co
浏览 149提问于2019-03-18得票数 2
回答已采纳
我在学习火星雨和mllib。
在使用RF模型预测测试数据之后，我将在一个名为“预测”的变量( RDD )中分配结果。
如果我调用predictions.count()或prediction.collect()，则在以下异常情况下，它将失败。
你能告诉我你的想法吗？已经花了不少时间，却找不到缺少的东西。
predictions = predict(training_data, test_data)
File "/mp5/part_d_poc.py", line 36, in predict
print(predictions.count())
Fil
浏览 0提问于2019-05-07得票数 2
1回答
根据值动态创建列
、、
在我们的项目中，我们使用一个ETL框架(内置在PySpark中)，它接受Spark-sql语句来转换来自源的数据并加载Hive外部表。 现在，我需要读取一个表，然后选择一列的唯一值，然后动态创建一个包含这么多列的Hive表。 例如，请考虑下表。 第一天： qstn_mstr： prsnid | qstn
1 | satisfactory
2 | survey
3 | review 目标表结构： prsnid | satisfactory | survey | review 在第1天:在上面的示例中，qstns列中有3唯一值，因此将使用这些3值作为列创建输出表。 第二天:说不。Qstn_m
浏览 14提问于2020-01-15得票数 0
我有以下序言代码，在我的两个场景之间共享：
from pyspark.sql import SparkSession
from pyspark.sql.types import *
import pyspark.sql.functions as F
import pandas as pd
import numpy as np
spark = SparkSession.builder.getOrCreate()
df = pd.DataFrame({"col1": [1, 2, 3], "col2": [22.0, 88.0, np.nan]
浏览 1提问于2022-01-03得票数 0
回答已采纳
我正在尝试理解DataFrame列类型。当然，DataFrame不是一个物化的对象，它只是一组Spark的指令，将来要转换成代码。但我认为，这个类型列表代表了在执行操作时JVM中可能出现的对象类型。
import pyspark
import pyspark.sql.types as T
import pyspark.sql.functions as F
data = [0, 3, 0, 4]
d = {}
d['DenseVector'] = pyspark.ml.linalg.DenseVector(data)
d['old_DenseVector'] =
浏览 1提问于2016-07-31得票数 7
回答已采纳
1回答
我手头有一个问题陈述，我想在spark-sql/pyspark中取消透视表。我已经看过文档了，我可以看到到目前为止只支持pivot，但没有支持support。有什么办法可以做到这一点吗？
让我的初始表看起来像这样：
当我使用下面提到的命令在pyspark中旋转它时：
df.groupBy("A").pivot("B").sum("C")
我将其作为输出：
现在我想取消旋转工作台的枢轴。一般来说，这个操作可能会/可能不会产生原始表，这取决于我如何透视原始表。
到目前为止，Spark-sql还没有提供对unpivot的开箱即用支持
浏览 3提问于2017-02-26得票数 23
回答已采纳
我正在尝试从pandas_udf返回一个特定的结构。它在一个集群上工作，但在另一个集群上失败。我尝试在组上运行udf，这需要返回类型为数据框架。
from pyspark.sql.functions import pandas_udf
import pandas as pd
import numpy as np
from pyspark.sql.types import *
schema = StructType([
StructField("Distance", FloatType()),
StructField("CarId", IntegerTy
浏览 2提问于2018-03-26得票数 5
回答已采纳
2回答
我使用以下代码动态过滤数据(我使用MySQL作为DB引擎)：
var filter = string.Empty;
if (!string.IsNullOrWhiteSpace(filterNumber))
{
filter = "t.Number LIKE :filterNumber";
}
var query = string.Format("SELECT t FROM Table t WHERE 1=1 AND {0} ORDER BY {1} {2}", filter, orderBy, orderDirection);
var q = Sess
浏览 0提问于2012-11-05得票数 0
回答已采纳
1回答
在pyspark中，我有以下几点： import pyspark.sql.functions as F
cc = F.lit(1).alias("A")
print(cc)
print(cc._jc.toString()) 我得到了： Column<b'1 AS `A`'>
1 AS `A` 有没有办法让我直接从cc打印"A“？看起来我不能很容易的提取别名。 我还认为，在scala中的spark-sql中，如果我打印"cc“，它将只打印"A”
浏览 40提问于2019-03-07得票数 0
3回答
这样做是可行的：
from pyspark.sql import SparkSession
from pyspark.sql.functions import pandas_udf
import pandas as pd
spark = SparkSession.builder.getOrCreate()
@pandas_udf(returnType="long")
def add_one(v: pd.Series) -> pd.Series:
return v.add(1)
spark.udf.register("add_one", add
浏览 10提问于2021-10-18得票数 6
回答已采纳
1回答
我想用regexp_replace替换Pyspark中字符串的一部分，比如“www.”。和“.com”。是否可以传递要替换的元素列表？
my_list = ['www.google.com', 'google.com','www.goole']
from pyspark.sql import Row
from pyspark.sql.functions import regexp_replace
df = sc.parallelize(my_list).map(lambda x: Row(url = x)).toDF()
df.withColum
浏览 14提问于2018-08-21得票数 1
回答已采纳
1回答
我最近开始学习用于网络图形操作的pySpark。我仍在试图理解星火如何使用数据格式。
据我所知：
dataframes中的数据是在一组机器上分布的(就像它们与hadoop一样)。
对dataframes的操作以相同的方式分布(或者，在我的例子中，在本地使用给定数量的线程)。
我的主要问题是:只有数据文件上的操作是分布式的，还是所有操作(比如python的字典上的操作)都是在pyspark的shell中执行后才分发的？
如有任何进一步的澄清或条款，将不胜感激。
浏览 1提问于2021-03-14得票数 1
我在pyspark SQL中的spark-sql中遇到以下查询转换问题-server查询是
coalesce((Select top 1 f2.ChargeAmt from Fact_CMCharges f2
where f2.BldgID = f.BldgID
and f2.LeaseID = f.LeaseID
and f2.IncomeCat = f.IncomeCat
and f2.Period < f.Period
and f2.ActualProjected = 'Lease'
浏览 1提问于2016-11-01得票数 1
在Spark集群上使用pyspark编程，数据量大且分片，因此不能轻松地加载到内存中或检查数据的健全性
基本上它看起来像是
af.b Current%20events 1 996
af.b Kategorie:Musiek 1 4468
af.b Spesiaal:RecentChangesLinked/Gebruikerbespreking:Freakazoid 1 5209
af.b Spesiaal:RecentChangesLinked/Sir_Arthur_Conan_Doyle 1 5214
维基百科数据：
我从亚马逊网络服务的S3上读到它，然后尝试用pyspark解释器中的以下p
浏览 0提问于2015-10-14得票数 11
回答已采纳
当我使用RegexTokenizer from pyspark.ml.feature来标记我的dataframe中的句子列以查找所有单词时，当python 包用于同一个句子时，我得到的结果正好相反。下面是示例代码：
from pyspark.sql import SparkSession
from pyspark.ml.feature import RegexTokenizer
spark = SparkSession.builder \
.master("local") \
.appName("Word list") \
浏览 3提问于2020-05-06得票数 1
回答已采纳
我使用spark-sql进行数据迁移项目。那么我应该如何在spark中实现stage area呢？何时使用spark sql缓存或持久化？有没有实时用例？
~Sha
浏览 0提问于2018-10-12得票数 0
1回答
我有一段pyspark代码，用于将数据帧转换为物理表：
df.write.mode('overwrite).saveAsTable('sometablename')
如果数据帧df包含名称中包含空格的列，则会失败，并显示以下错误：
18/03/08 10:33:29 ERROR CreateDataSourceTableAsSelectCommand: Failed to write to table pivot_up_spaces_Export_Data_4
org.apache.spark.sql.AnalysisException: Attribute name
浏览 4提问于2018-03-08得票数 1
当使用Python中的Polars读取CSV文件时，我们可以使用参数dtypes来指定要使用的模式(对于某些列)。我想知道我们在读或写Parquet文件时能做同样的事情吗？我试图指定dtypes参数，但它不起作用。
我有一些从PySpark生成的Parquet文件，并希望将这些Parquet文件加载到Rust中。锈蚀需要无符号整数，而火花/PySpark没有无符号整数，并将带符号整数输出到Parquet文件中。为了使事情更简单，我想在将Parquet文件加载到Rust之前转换它们的类型。我知道要做到这一点有几种不同的方法(包括熊猫和北极熊)，但我不知道是否有简单有效的方法使用极地来实现这一目标
浏览 13提问于2022-05-20得票数 1
我尝试在单个Pyspark转换中从单个目录读取(所有或多个)数据集。有没有可能迭代路径中的所有数据集，而不将单个数据集硬编码为输入？ 我希望动态地从多个数据集中获取不同的列，而不必对单个输入数据集进行硬编码。
浏览 24提问于2020-09-22得票数 0
回答已采纳
2回答
我尝试使用Pyspark将不同的拼图文件读取到一个数据帧中，但它给了我错误，因为多个拼图文件中的一些列具有不同数据类型的列。例如:列"geo“在某些文件中的数据类型为"Double”，在另一些文件中为"String“。
我应该如何处理这个问题？我必须在多个拼图文件中手动转换具有不同数据类型的每一列吗？
浏览 12提问于2021-11-22得票数 0
我试图使用PySpark读取包含多列的CSV文件。inferschema选项非常适合推断大多数列的数据类型。如果我只想覆盖被错误推断的列类型之一，那么最好的方法是什么？
我可以使用这段代码，但它使PySpark只导入了模式中指定的一个列，这是我不想要的。
schema = StructType() \
.add("column_one_of_many", StringType(), True)
spark.read.format('com.databricks.spark.csv') \
.option('delimited'
浏览 1提问于2021-08-27得票数 2
回答已采纳
我对Spark很陌生，并试图接受它。我们使用Spark-SQL主要用于ETL的数据转换/操作。最近，我偶然发现了一些用于操作星火复杂类型的SparkSQL函数(array、map、struct等)。我遇到了某些函数，如array()、array_zip()、struct()、map_concat()、map_from_arrays()等。我在网上搜索，但不幸的是，我无法获得足够的示例/解释，让我对它们有一个清晰的理解。
请任何人提供一些它的应用程序的例子，这样我就能更好地理解它们在我们的项目任务中的应用。
注意:我只能访问Spark-SQL，但不能访问 PySpark-SQL。
谢谢
浏览 1提问于2020-05-31得票数 0
1回答
我对pyspark是个新手，我正面临着regex的问题。因此，我有一个日期列，我想检查列中的行是否为日期格式 ..。我知道如何在普通的python中做到这一点，但不知道pyspark，有人能帮助我吗？ 我的代码： df =
Date name
0 12/12/2020 a
1 24/01/2019 b
2 08/09/2018 c
3 12/24/2020 d
print(re.search('\d.*\/.*\d.*\/.*\d', str(df['Date']))) 这提供了：
浏览 54提问于2021-02-28得票数 0
回答已采纳
1回答
col1 col2
0 2017年-02-04 10:41:00.0000000
1 B 2017-02-04 10:41:00.0000000
2 C 2017-02-04 10:41:00.0000000
3 D 2017-02-04 10:41:00.0000000
4 E 2017-02-03 06:13:00.0000000
我有多个，所有的列数据类型都是字符串。我想过滤掉那些像模式一样有日期时间的列名。假设上面给出了这个，并将所有列的数据类型作为字符串。我想要编写一段代码，自动检测具有日期时间格式值的列。因此，在上面的dataframe中，它应该返回col2作为输出。我在Pyt
浏览 10提问于2021-06-01得票数 0
1回答
我正在学习PySpark。在中，有一个例子：
from pyspark.ml.linalg import Vectors
from pyspark.ml.classification import LogisticRegression
# Prepare training data from a list of (label, features) tuples.
training = spark.createDataFrame([
(1.0, Vectors.dense([0.0, 1.1, 0.1])),
(0.0, Vectors.dense([2.0, 1.0, -1.
浏览 0提问于2019-02-13得票数 1
回答已采纳
我正在尝试找出spark数据框中的列是什么数据类型，并基于该定义操作列。
这是我到目前为止所知道的：
import pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('MyApp').getOrCreate()
df = spark.read.csv('Path To csv File',inferSchema=True,header=True)
for x in df.columns:
if type(x) == 'integer
浏览 8提问于2018-04-12得票数 1
我想使用包含关键字的字典，并检查pyspark df中的一列，看看该关键字是否存在，如果存在，则在新列中返回字典中的值。
问题看起来是这样的；
myDict = {
'price': 'Pricing Issue',
'support': 'Support Issue',
'android': 'Left for Competitor'
}
df = sc.parallelize([('1','Needed better Support'),('2'
浏览 0提问于2021-03-29得票数 0
如何向现有表中插入新数据？
我正在尝试使用pyspark将新数据插入到现有表中。
这是我的节目
from pyspark import SparkContext
from pyspark.sql import SQLContext, DataFrameWriter
sc = SparkContext("local[*]", "SPARK-SQL")
sqlContext = SQLContext(sc)
df = sqlContext.read.json("people.json")
df.registerTempTable("peo
浏览 0提问于2018-11-08得票数 3
3回答
我安装了anaconda，还下载了Spark 1.6.2。我正在使用这个答案中的以下说明来为Jupyter 配置spark
我已经将spark目录下载并解压缩为
~/spark
现在，当我cd进入这个目录和bin时，我看到了以下内容
SFOM00618927A:spark $ cd bin
SFOM00618927A:bin $ ls
beeline pyspark run-example.cmd spark-class2.cmd spark-sql sparkR
beeline.cmd pyspark.cmd run-
浏览 4提问于2016-08-06得票数 6
在PySpark中，我们不能使用传统的熊猫或R风格的符号来根据DataFrame中的旧列来制作新的列。例如，假设我试图连接两列：
import pyspark.sql.functions as F
df['newcol'] = F.concat(df['col_1'], df['col_2'])
结果：
TypeError: 'DataFrame' object does not support item assignment
相反，这样做的实现方法是
df = df.withColumn('newcol', F.
浏览 1提问于2016-07-28得票数 1
回答已采纳
当我试图在Spark-SQL外部数据源上做一些测试时，会发生这个问题。
我以两种方式构建数据帧，并比较收集操作的速度。而且我发现如果列数太大，从外部数据源构建的数据帧将会滞后。我想知道这是否是Spark-SQL的外部数据源的限制。:-)
为了更清楚地表达这个问题，我编写了一段代码：
在我的外部Datasource API基准代码中，它实现了一个假的外部数据源(实际上是一个RDD[String，ArrayInt] )，并通过以下方式获取数据帧
val cmpdf = sqlContext.load("com.redislabs.test.dataframeRP", Map[St
浏览 2提问于2016-01-09得票数 1
1回答
首先，请允许我先说一句，我对Spark-SQL非常陌生。 我正在尝试理解Spark-Sql中的各种连接类型和策略，我希望能够了解一种近似表大小(参与连接、聚合等)的方法，以便通过了解幕后实际发生的事情来估计/调整预期的执行时间，以帮助我选择最适合该场景的连接策略(在Spark-SQL中通过提示等)。 当然，表行计数提供了一个很好的起点，但我希望能够估计bytes/KB/MB/GB/TBs，的大小，以便了解哪个表可以/不可以放入内存等)，这反过来将允许我通过选择最适合该场景的连接类型/策略等来编写更有效的SQL查询。 注意:我没有访问PySpark的权限。我们通过使用Hive jdbc驱动程序
浏览 106提问于2020-06-10得票数 2
在pyspark中创建示例数据
from pyspark.sql.types import *
from pyspark.sql.types import StructField
from pyspark.sql import types
testdata = [("aaaa",1,50.0,"05-APR-2020"),
("bbbb",2,100.0,"06-APR-2020")]
dataschema = types.StructType([
typ
浏览 1提问于2020-05-04得票数 0
回答已采纳
3回答
有关火花提交的文件如下：
Spark的bin目录中的星火提交脚本用于启动集群上的应用程序。
关于火花放电，它说：
您还可以使用bin/pyspark启动交互式Python。
这个问题听起来很愚蠢，但是当我通过pyspark运行命令时，它们也运行在“集群”上，对吗？它们不只是在主节点上运行，对吗？
浏览 0提问于2016-04-28得票数 4
回答已采纳
我想在上的笔记本中使用PySpark创建一个简单的数据格式。dataframe只有3列：
stringStartTimeStanp - 'HH:MM:SS:MI'*EndTimeStanp -数据类型，例如“时间戳”或可以在表单‘HH:MM:SS:MI’*EndTimeStanp中持有时间戳(无日期部分)的数据类型--类似于“时间戳”之类的数据类型或可以在表单'HH:MM:SS:MI'*中持有时间戳(无日期部分)的数据类型。
*小时:分钟:秒:毫秒，例如“15:59:59:59:59”
Example values for one row:
TimePeriod
浏览 2提问于2021-06-22得票数 4
回答已采纳
如何创建像https://docs.databricks.com/spark/latest/spark-sql/language-manual/create-function.html#create-function这样函数，但在python中定义该函数？ 我已经这样做了： from pyspark.sql.types import IntegerType
def relative_month(input_date):
if input_date is not None:
return ((input_date.month + 2) % 6)+1
else:
retu
浏览 31提问于2019-03-02得票数 2
1回答
我尝试从PySpark中的REST中提取不同的表。我跟踪了这个。我想要将不同的模式存储在一列中的中。下面是一个示例：
import pyspark.sql.functions as F
from pyspark.sql import Row
from pyspark.sql.types import *
A = [{"TableName": "Table1", "Schema": StructType([StructField("a", StringType()), StructField("b", Intege
浏览 16提问于2022-08-23得票数 0
回答已采纳
我正在尝试将pandas.DataFrame代码转换为等效的pyspark DataFrame。我有一个以下格式的RDD。
myRdd = [[1, 'a', {'a':[1, 2]}],
[2, 'b', {'c': 1, 'd':3}],
[3, 'c', {}]]
columnNames = ['sl', 'name', 'params']
内部列表中的第三个元素没有特定的结构。在pandas数据帧中，我可以
浏览 1提问于2018-04-07得票数 1
SampleCSV2.csv的数据如下：
AAA|25|IT|50.5
BBB|28|Comp|100.5
在pyspark中创建具有特定数据类型的dataframe时遇到问题
from pyspark import SparkContext, SparkConf, SQLContext
from pyspark.sql.types import StructType, StructField, DoubleType, IntegerType, StringType, DateType, TimestampType
from pyspark.sql.types import *
def s
浏览 6提问于2017-09-04得票数 1
回答已采纳
我有一个用例来映射基于条件的pyspark列的元素。通过这个文档，我找不到一个函数来执行映射函数。
因此，尝试使用pyspark映射函数，但无法将pyspark列转换为dataFrame
注意:我之所以使用pyspark列，是因为我从我使用的库(远大期望)中获得了它的输入。
@column_condition_partial(engine=SparkDFExecutionEngine)
def _spark(cls, column, ts_formats, **kwargs):
return column.isin([3])
# need to replace the abov
浏览 4提问于2021-11-19得票数 0
回答已采纳
1回答
我从SEC Edgar ()下载了一份上市公司名单。
我正在尝试查找不同列表中的公司之间的匹配，这些公司在大小写和标点符号方面可能会有所不同。对于在两列之间匹配公司名称有什么建议吗？
下面是一个例子：
> df.a
Company
"Flowers David"
"Danone Inc."
> df.b
Company
"1 800 FLOWERS COM INC"
"FLOWERS DAVID J A"
"FLOWERS FOODS INC"
我正在尝试以一种有意义的方式在两者之间找到匹配
浏览 1提问于2014-06-27得票数 0
我将使用PySpark运行一些东西(回归、ML等)。在相当大的结构化数值数据集(>1TB)上。目前，谷歌云存储上的数据是CSV格式的。是否将数据转换为其他格式(Parquet等)？在csv中加速阅读火花，还是阅读速度大致相同？
浏览 2提问于2017-06-27得票数 0
回答已采纳
我需要在PySpark中定义元数据。有些列具有字符串类型的最大长度。我注意到在中有VarcharType类型。然而，它在pyspark.sql.types中并不存在。
from pyspark.sql.types import *
my_schema = StructType([
StructField("POSTAL_CODE", VarcharType(4))
, StructField("CITY", VarcharType(20))
])
NameError: name 'VarcharType' is not defin
浏览 6提问于2022-08-30得票数 0
回答已采纳
1回答
我正在编写一个正则表达式，以应用于PySpark DataFrame列。
我不可能碰巧在PySpark中复制我的工作正则表达式的结果。
我尝试过几种方法(见下文)，但似乎没有一种方法起作用(甚至在特定的Java regex引擎上也没有测试)。我想取上面的例子中显示的组。
(\w+(?:\s*|\d*)\s+RUE\s.*)
[\s\-]+(\d*\s*RUE\s+.*)
代码示例：
df = spark.createDataFrame([
('RESIDENCE LA VENDEENNE 80 81 RUE LOUIS LUMIERE',)
], ["adres
浏览 2提问于2019-11-20得票数 1
回答已采纳
如果我有一个包含arrays的Spark，我可以通过DataFrame在这些数组上使用Python List方法吗？如何将Spark DataFrame array<double>转换为Python列表？
下面是一个示例，其中包含一些UDF。我不确定为什么使用最大值可以，但是使用len就不行了。最后，我想用原始数组列中的采样值创建一个新列。这也会得到一个关于期望两个参数的错误，如果你能帮助的话也会得到加分！
我有以下Spark DataFrame
from pyspark.sql.functions import udf
from pyspark.sql import Row
fr
浏览 1提问于2016-10-28得票数 4
回答已采纳
我有这个错误发生，我可以在这个错误上找到多个堆栈溢出帖子，并且我能够将在那些posts.But中给出的解释联系起来--我的场景是不同的。我正在Spark1.6.1环境中执行下面的python代码：：
1 from pyspark import SparkContext
2 sc = SparkContext("local", "Simple App")
3 from pyspark.sql import SQLContext, Row
4 sqlContext = SQLContext(sc)
5
6 ssn = sc.textFile(
浏览 1提问于2017-05-24得票数 0
回答已采纳
3回答
如何将作为字符串读取的列转换为数组列？即从下面的模式转换
scala> test.printSchema
root
|-- a: long (nullable = true)
|-- b: string (nullable = true)
+---+---+
| a| b|
+---+---+
| 1|2,3|
+---+---+
| 2|4,5|
+---+---+
至：
scala> test1.printSchema
root
|-- a: long (nullable = true)
|-- b: array (nullable = true)
|
浏览 1提问于2017-06-22得票数 20
回答已采纳
我想加载数据到Hbase表使用pyspark，有人可以帮助如何加载的json数据到Hbase的行键作为和所有其他进入一个列族。请在下面找到json。
{ "ticid"："1496"，"ticlocation"："vizag"，"custnum"："222"，"Comments"：{ "comment"：{ "commentno"："1"，"desc"：“旅程”，“乘客座位”：{ "intele"：
浏览 1提问于2018-05-24得票数 0
假设我在python中创建了一个函数，所以将一个数字求幂为2： def squared(s):
return s * s 然后我在Spark session中注册了这个函数，如下所示： spark.udf.register("squaredWithPython", squared) 然后当我在Spark SQL中调用UDF时，如下所示： spark.range(1, 20).registerTempTable("test")
%sql select id, squaredWithPython(id) as id_squared from test 那么，如
浏览 33提问于2019-10-10得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
云直播
对象存储活动推荐
腾讯云开发者
