在spark中使用验证转换Dataframe列
在Spark中使用验证转换Dataframe列，可以通过使用Spark SQL提供的内置函数和表达式来实现。以下是完善且全面的答案：
验证转换（Validation Transformation）是指对Dataframe列进行验证和转换操作，以确保数据的准确性和一致性。在Spark中，可以使用内置函数和表达式来实现验证转换。
- 验证转换的概念：验证转换是指对数据进行验证和转换的过程，可以用来处理数据中的异常、缺失、错误等情况，以确保数据的准确性和一致性。
- 验证转换的分类：验证转换可以分为以下几种类型：
- 类型转换：将列的数据类型转换为所需的类型，例如将字符串转换为整数。
- 值的验证：对列中的值进行验证，例如检查是否为正数或是否满足某种模式。
- 缺失值处理：处理列中的缺失值，例如填充缺失值或删除包含缺失值的行。
- 数据清洗：对列中的数据进行清洗，例如去除空格、去重等操作。
- 数据修正：对列中的数据进行修正，例如修正日期格式或修正错误的数据。
- 验证转换的优势：
- 数据准确性：通过验证转换，可以确保数据的准确性和一致性，减少错误和异常的出现。
- 数据完整性：通过处理缺失值和清洗数据，可以提高数据的完整性。
- 数据一致性：通过类型转换和数据修正，可以提高数据的一致性和可用性。
- 验证转换的应用场景：
- 数据清洗和预处理：在数据分析和机器学习任务中，对原始数据进行清洗和预处理是非常重要的步骤。可以使用验证转换来处理缺失值、异常值、重复值等，提高数据质量。
- 数据规范化：在数据集成和数据交换中，将不同来源的数据进行规范化是必要的。可以使用验证转换来统一数据类型、格式和结构，提高数据的一致性。
- 数据修复和纠错：在数据处理和数据传输过程中，可能会出现一些错误和异常情况，需要对数据进行修复和纠错。可以使用验证转换来修正错误的数据、纠正格式不正确的数据等。
- 腾讯云相关产品和产品介绍链接地址：
- 腾讯云Spark：腾讯云提供的大数据处理和分析服务，支持使用Spark进行数据处理和计算。了解更多信息，请访问：https://cloud.tencent.com/product/emr-spark
- 腾讯云数据仓库（Cloud Data Warehouse）：腾讯云提供的数据仓库解决方案，支持数据集成、清洗、转换等功能。了解更多信息，请访问：https://cloud.tencent.com/product/dws
通过使用Spark中的内置函数和表达式，可以轻松实现验证转换操作，提高数据的质量和可用性。
相关·内容
1回答
我有一个从CSV文件导入的spark。在应用了一些操作(主要是删除列/行)之后，我尝试将新的DataFrame保存到Hadoop，这将显示一个错误消息：
ValueError:超出范围的年份
我怀疑DateType或TimestampType类型的某些列已损坏。至少在一篇专栏中，我找到了一篇带有'207‘年的条目--这似乎产生了一些问题。
**如何检查DataFrame是否遵守所需的时间范围？
我考虑编写一个函数，它接受DataFrame，并为每个DateType /TimestampType列获取值的最小值和最大值，但我无法让它工作。
有什么想法吗？
PS:据我理解，spar
浏览 1提问于2018-08-27得票数 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
回答已采纳
我试图使用Pyspark在dataframe中更改一个列，也就是跨行对单个列的所有值进行洗牌。
我试图避免这样的解决方案，即在将列拆分并分配索引列之前，将其重新连接到原始的dataframe，而原始dataframe也有一个添加的索引列。主要是因为我的理解(这可能是非常错误的)，在大型数据集(数百万行)的运行时，联接是不好的。
# for some dataframe spark_df
new_df = spark_df.select(colname).sort(colname)
new_df.show() # column values sorted nicely
spark_df.with
浏览 0提问于2019-06-06得票数 0
我有一个包含许多列的spark.ml DataFrame，每列中的每一行都包含一个SparseVector。我想对每一列应用MultivariateStatisticalSummary.colStats，colStats签名是：
def colStats(X: RDD[Vector]): MultivariateStatisticalSummary
看起来很完美..。除了我似乎不能从该DataFrame中select一列并将其转换为RDD[Vector]。这是我的尝试：
val df: DataFrame = data.select(shardId)
val col = df.as[(org
浏览 0提问于2017-02-16得票数 0
当创建一个列值默认值为NULL的表时，我在Spark(1.6) SQL中获得了下面的错误。示例:将表测试创建为select column_a，NULL作为column_b从test_temp创建；
同样的东西在Hive中工作，并创建数据类型为"void“的列。
我使用空字符串而不是NULL来避免异常和新列获取字符串数据类型。
是否有更好的方法使用spark在hive表中插入空值？
2017-12-26 07:27:59 ERROR StandardImsLogger$:177 - org.apache.hadoop.hive.ql.metadata.HiveException: jav
浏览 3提问于2017-12-27得票数 0
回答已采纳
1回答
当在Spark Dataframe上的过滤器中使用"and“子句时，它返回Spark.SQL.Column而不是Spark Dataframe。但在一个条件下，它工作得很好。
如何show()或迭代通过Spark Sql列对象？尝试show()函数时抛出错误- 'Column' object not callable.
或者如何将Spark.SQL.Column转换为Spark Dataframe？或者如何在filter子句中传递多个条件？
df2 = df.filter((df.dropoff_longitude >= -65.09)&(df.dropo
浏览 1提问于2019-04-23得票数 0
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
回答已采纳
1回答
我想把数据库中的org.apache.spark.sql.DataFrame 转换成 org.apache.spark.rdd.RDD[(String, String)] ，有人能帮上忙吗？
背景(更好的解决方案也是受欢迎的)：我有一个Kafka流，它(经过一些步骤)变成了一个2列数据框架。我想把它放到Redis缓存中，第一列作为键，第二列作为值。
更具体地说，输入的类型是：lastContacts: org.apache.spark.sql.DataFrame = [serialNumber: string, lastModified: bigint]。我试着按以下方式对Redis进行分析：
浏览 0提问于2019-03-25得票数 0
回答已采纳
我正在使用下面的语句在spark中读取csv。
df = spark.read.csv('<CSV FILE>', header=True, inferSchema = True)
当我检入spark dataframe时，一些整型和双精度列被存储为dataframe中的字符串列。但是，并不是所有的列都是这样。
我已经检查了特定列的值，所有的值都是双精度类型，但spark仍然推断为StringType。
因为我加载的CSV文件大约有1000列，所以也显式地指定模式是不可行的。
如有任何建议或帮助，我们将不胜感激。
致以敬意，
Neeraj
浏览 5提问于2017-08-31得票数 2
1回答
我想在我的星火程序中遍历dataframe列，并计算min和max值。我对Spark和scala很陌生，一旦我在dataframe中获取它，我就无法遍历这些列。
我尝试过运行以下代码，但它需要将列号传递给它，问题是如何从dataframe中获取它并动态传递它，并将结果存储在一个集合中。
val parquetRDD = spark.read.parquet("filename.parquet")
parquetRDD.collect.foreach ({ i => parquetRDD_subset.agg(max(parquetRDD(parquetRDD.colum
浏览 3提问于2017-07-18得票数 0
回答已采纳
我有一个有bigint列的Dataframe。如何将bigint列转换为scala spark中的时间戳
浏览 1提问于2019-07-23得票数 4
回答已采纳
在我的AWS集群中，我下载了一个Python包：
python3 -m pip install Unidecode
现在，我想在我的名为“'city_no_accents‘city”的pyspark列中使用这个词，该列的值为：’s o paulo‘、'seropédica’等(即带有口音)，并希望创建一个名为“”的新列，它将更正文本中的所有重音，并使其在普通英语文本中，如‘圣保罗’、'seropedica‘等等。
因此，我编写了下面的PySpark代码：
<...imported some other packages>
from unidecode i
浏览 5提问于2020-10-15得票数 0
1回答
在Spark1.6.0中，我有一个包含职务描述的列的数据框架，如：
Description
bartender
bartender
employee
taxi-driver
...
我使用以下方法从该列检索唯一值列表：
val jobs = people.select("Description").distinct().rdd.map(r => r(0).asInstanceOf[String]).repartition(4)
然后，对于每个职务描述，我尝试检索具有该职务的人员并做一些事情，但我得到了一个NullPointerException：
jobs.foreach
浏览 2提问于2016-02-21得票数 5
回答已采纳
在尝试将cassandra数据加载到一个writeTime (DataFrame)中时，我试图弄清楚是否可以使用火花卡桑德拉连接器获取列的DataFrame。
这就是我想要做的：
val df = spark.read.format("org.apache.spark.sql.cassandra")
.options(Map( "table" -> "table1", "keyspace" -> "keyspace1",
"cluster" ->
浏览 0提问于2018-03-14得票数 1
回答已采纳
有没有一种方法可以使用StructType转换数据帧的所有值？
让我用一个例子来解释我的问题：
假设我们从文件读取后获得了一个数据帧(我提供了一个生成此数据帧的代码，但在我的真实项目中，我是在从文件读取后获得此数据帧的)：
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._
import spark.implicits._
val rows1 = Seq(
浏览 7提问于2018-07-28得票数 4
回答已采纳
在所有发布的关于这个行动的问题中，我找不到有用的东西。
我尝试了几个版本，在所有的版本中，我都有一个DataFrame
dataFrame = spark.read.format("com.mongodb.spark.sql").load()
dataFrame.printSchema()的打印输出是
root
|-- SensorId: string (nullable = true)
|-- _id: struct (nullable = true)
| |-- oid: string (nullable = true)
|-- _type: string (
浏览 1提问于2017-10-25得票数 3
回答已采纳
我在Apache Spark SQL中有一个DataFrame，我想删除所有not None值都相同的列。 所以在一个虚拟的例子中 df
| A | B | C |
1 2 3
NaN 2 4
1 2 NaN
1 2 5 我只想保留C列 df_filter
| C |
3
4
NaN
5 在Python中，我将通过以下方式完
浏览 25提问于2021-10-27得票数 2
回答已采纳
我通过拼图文件将大量数据读入到数据帧中。我注意到大量的列都有1,0，-1作为值，因此可以从Int类型转换为Byte类型，以节省内存。
我写了一个函数来做这件事，并返回一个新的dataframe，其中的值被转换为字节，但是当在UI中查看dataframe的内存时，我发现它只是保存为原始dataframe的转换，而不是新的dataframe本身，因此占用了相同的内存量。
我是Spark的新手，可能不完全理解ByteType的内部原理，那么我应该如何开始将这些列设置为Spark的？
浏览 3提问于2018-02-01得票数 5
1回答
我正在尝试访问一个hive表，并从表/ dataframe中提取和转换某些列，然后将这些新列放到一个新的dataframe中。我正试着这样做-
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)
val hiveDF = sqlContext.sql("select * from table_x")
val system_generated_id = hiveDF("unique_key")
val application_assigned_event_id = hiveDF("
浏览 1提问于2016-07-03得票数 1
回答已采纳
我有一个从熊猫数据帧创建的样本spark数据帧-
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql.types import StringType
from pyspark.sql.types import *
import pandas as pd
spark = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.config(&
浏览 3提问于2020-09-09得票数 0
我有一个包含布尔列(TRUE、FALSE、NULL)的dataframe。当我使用Spark2 Scala将此数据发送到Oracle19c中的现有表时，由于错误: java.sql.SQLException:无效列类型: 16，它失败了。
但是，当它们不是dataframe中的空值时，任务就成功地完成了。此外，如果我将一个空值直接插入到Oracle中的布尔字段中，我就没有问题(这只是为了在Oracle中直接测试空值)。
我使用Spark在Oracle中编写数据的方式如下：
df.write.mode(SaveMode.Append).jdbc(url, tableName, connectio
浏览 2提问于2021-03-21得票数 1
我应该使用python中的哪个.map()函数从spark数据帧创建一组labeledPoints？如果标签/结果不是第一列，但我可以引用它的列名'status‘，那么表示法是什么？
我使用这个.map()函数创建Python dataframe：
def parsePoint(line):
listmp = list(line.split('\t'))
dataframe = pd.DataFrame(pd.get_dummies(listmp[1:]).sum()).transpose()
dataframe.insert(0, 's
浏览 3提问于2015-09-14得票数 14
回答已采纳
1回答
我是新来的火花/斯卡拉。我正在尝试读取一些数据从一个蜂窝表到一个火花数据，然后添加一个列的基础上的一些条件。这是我的代码：
val DF = hiveContext.sql("select * from (select * from test_table where partition_date='2017-11-22') a JOIN (select max(id) as bid from test_table where partition_date='2017-11-22' group by at_id) b ON a.id=b.bid")
浏览 1提问于2017-11-27得票数 0
回答已采纳
我有一个包含日期时间列的csv文件："2011-05-02T04:52:09+00:00“。
我使用scala，文件被加载到spark DataFrame中，我可以使用jodas时间来解析日期：
val sqlContext = new SQLContext(sc)
import sqlContext.implicits._
val df = new SQLContext(sc).load("com.databricks.spark.csv", Map("path" -> "data.csv", "header" -
浏览 0提问于2015-04-28得票数 3
回答已采纳
我在Scala上创建一个Session，它将大量数据从MongoDB中提取到DataFrame中。每当我试图只显示该DataFrame的一列时，就会得到以下错误：
(run-main-0) java.lang.IllegalStateException: SparkContext has been shutdown
我已经尝试将数据的大小缩小到只有10行，但是仍然会出现相同的错误。最终目标是将"Rev“列中的所有值求和并作为变量返回，但首先需要克服这个问题。
下面是我的代码片段，其中URI
val sparkMongo = SparkSession.builder() // Creat
浏览 3提问于2020-11-04得票数 0
5回答
如何使用spark-ml而不是spark-mllib处理分类数据？
虽然文档不是很清楚，但似乎分类器(例如RandomForestClassifier、LogisticRegression )有一个featuresCol参数，它指定了DataFrame中的特性列的名称，还有一个labelCol参数，它指定了DataFrame中标记类的列的名称。
显然，我希望在我的预测中使用多个特性，所以我尝试使用VectorAssembler将所有特性放在featuresCol下的一个向量中。
但是，VectorAssembler只接受数字类型、布尔类型和向量类型(根据星火网站)，所以我不能在特性向量中放置字
浏览 10提问于2015-08-28得票数 49
我需要在其中一列中保存包含org.apache.spark.mllib.linalg.Vector的dataframe。在cassandra中从dataframe创建表时，抛出错误。
java.lang.Exception: Cassandra Writer Failed. java.lang.IllegalArgumentException: Unsupported type: org.apache.spark.mllib.linalg.VectorUDT@f71b0bce
at com.datastax.spark.connector.types.ColumnType$.u
浏览 28提问于2016-08-09得票数 3
我想把Spark dataframe所有的时间戳列转换成字符串列。有人能告诉我如何为每个数据帧自动做到这一点吗？
时间戳列的位置可以改变，每个数据帧的列名也可以不同。
例如，在其中一个数据帧中，它可以是DataFrame1 columnA，但在Dataframe2中，它可以是columnX。
因此，我需要在任何给定表中使用有关列类型的信息，并将其转换为字符串列。
你有什么想法吗？
浏览 66提问于2019-11-08得票数 1
在我的理解中，列格式更适合于MapReduce任务。即使对于某些列的选择，columnar也能很好地工作，因为我们不必将其他列加载到内存中。
但是在Spark3.0中，我看到在查询计划中应用了这个ColumnarToRow操作，根据我从可以理解的，查询计划将数据转换为行格式。
它如何比柱状表示更有效，管理该规则应用的洞察力是什么？
对于下面的代码，我附加了查询计划。
import pandas as pd
df = pd.DataFrame({
'a': [i for i in range(2000)],
'b': [i for i in re
浏览 13提问于2020-11-12得票数 10
我正在尝试执行星火的countVectorizer模型。作为这个需求的一部分，我正在读取一个csv文件并从其中创建一个Dataframe (inp_DF)。
它有3列，如下所示
+--------------+--------+-------+
| State|Zip Code|Country|
+--------------+--------+-------+
| kentucky| 40205| us|
| indiana| 47305| us|
|greater london| sw15| gb|
| ca
浏览 1提问于2017-09-05得票数 0
回答已采纳
我用Python训练了一个随机森林算法，并希望用PySpark在一个大数据集上应用它。
我首先加载了经过训练的sklearn模型(使用joblib)，将包含这些特性的数据加载到Spark数据框架中，然后添加了一个带有预测的列，该列具有用户定义的函数，如下所示：
def predictClass(features):
return rf.predict(features)
udfFunction = udf(predictClass, StringType())
new_dataframe = dataframe.withColumn('prediction',
udf
浏览 6提问于2017-05-31得票数 8
回答已采纳
10回答
我有一个火花DataFrame (使用PySpark 1.5.1)，并希望添加一个新的列。
我尝试过以下几种方法，但都没有成功：
type(randomed_hours) # => list
# Create in Python and transform to RDD
new_col = pd.DataFrame(randomed_hours, columns=['new_col'])
spark_new_col = sqlContext.createDataFrame(new_col)
my_df_spark.withColumn("hours"
浏览 13提问于2015-11-12得票数 179
回答已采纳
我目前正在尝试学习在Scala中使用Apache Spark。 我有下表作为我想要用于分析的数据框架 ? 现在我要遍历各行，获取body列中字符串的id和字数，并将信息输出到包含2列的dataframe中。 def analyseDF(df:DataFrame): Unit = {
//var retFrame = spark.emptyDataset[ClassIdCount].toDF()
var tList = mutable.MutableList[IdCount]()
df.foreach(row => {
val
浏览 28提问于2019-02-27得票数 0
我准备使用spark dataframe (不是pandas)的内置CSV打印工具。我已经构建了一个IndexedRowMatrix。因此，它中有稀疏数组列。到密集数组的转换是由x.vector.toArray()映射调用执行的(如下所示)。我似乎想不出，如何将密集的numpy数组元素放入Spark DataFrame的单独列中。(我不是在介绍熊猫。)如何将此RDD转换为7列数据帧，由一个字符串列和6个整数列组成？到目前为止我的代码如下：
X = CoordinateMatrix(sc.parallelize(entries))
Xirm = X.toIndexedRowMatrix()
c
浏览 0提问于2018-06-12得票数 0
2回答
将星火数据转换为矢量
、、、
我希望使用朴素分类器模型来预测星火dataframe的输出类。我使用Spark2.1.0的结构化流功能。
当我尝试这样做的时候：
tokenizer = Tokenizer(inputCol="message", outputCol="logTokenize")
tokenizeData = tokenizer.transform(stream_df)
hashingTF = HashingTF(inputCol="logTokenize", outputCol="rawFeatures", numFeatures = 1000
浏览 5提问于2017-07-24得票数 0
1回答
我在Spark中有一个dataframe (通过读取csv创建)，如何在C#中遍历这个数据文件中的行。dataframe中有10行3列，在逐行导航时，我希望获得每个列的值。以下是我正在尝试的：
foreach (var obj in df)
{
Console.WriteLine("test");
}
foreach语句不能对“DataFrame”类型的变量进行操作，因为“DataFrame”不包含“GetEnumerator”的公共实例定义
浏览 4提问于2022-11-09得票数 0
回答已采纳
我有一个从sql上下文构建的spark dataframe。我使用DATE_FORMAT(time, 'Y/M/d HH:00:00') AS time_hourly截断了a日期时间字段
现在，列类型是一个字符串。如何将字符串dataFrame列转换为datetime类型？
浏览 2提问于2016-08-18得票数 1
我需要根据数据类型过滤Spark dataFrame列中的值。我想在一列中只有浮点数。我尝试使用一些正则表达式，但是，在写入csv文件时出现错误：SparkException: Task not serializable
下面是从CSV文件读取到dataFrame的方法，然后我过滤一些列并将它们写回csv文件：
def processDatasetCsvWithSpark(sqlContext: SQLContext, columnNames: Seq[String], filename: String, dfSchema: StructType,
浏览 0提问于2018-04-02得票数 0
我有一个dataframe：yearDF，它是通过读取Postgres上的关系型数据库管理系统表获得的，我需要在HDFS上的Hive表中摄取它。
val yearDF = spark.read.format("jdbc").option("url", connectionUrl)
.option("dbtable", s"(${execQuery}) as year2017")
.option("user", d
浏览 9提问于2018-08-30得票数 0
在spark批处理作业中，我通常会将JSON数据源写入到一个文件中，并且可以使用DataFrame阅读器的损坏列功能将损坏的数据写出到不同的位置，并使用另一个阅读器从同一作业中写入有效数据。(数据写为拼接) 但在Spark Structred Streaming中，我首先通过kafka将流作为字符串读取，然后使用from_json获取我的DataFrame。然后，from_json使用JsonToStructs，它在解析器中使用FailFast模式，并且不将未解析的字符串返回到DataFrame中的列。(参见参考文献中的注释)那么我如何使用SSS将与我的模式不匹配的损坏数据和可能无效的JSO
浏览 34提问于2018-12-27得票数 1
回答已采纳
3回答
我使用的是Spark 1.3.1。
我正在尝试查看Python中Spark dataframe列的值。有了Spark dataframe，我可以使用df.collect()来查看数据帧的内容，但在我看来，Spark dataframe列还没有这样的方法。
例如，数据帧df包含一个名为'zip_code'的列。所以我可以做df['zip_code']，它会变成一个pyspark.sql.dataframe.Column类型，但是我找不到一种方法来查看df['zip_code']中的值。
浏览 1提问于2015-06-30得票数 43
回答已采纳
我已经从np.select获得了一个numpy数组，我希望将其存储为PySpark DataFrame中的一个新列。我怎么能这么做？
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
pdf = pd.DataFrame({'a': [1,2,3], 'b': ['abc', 'cde', 'edf']})
df_data = spark.createDataFrame(pdf, schema='
浏览 6提问于2022-05-25得票数 1
1回答
我正在dataframe1中读取CSV文件，然后在dataframe2中筛选一些列，在从dataframe1中选择dataframe2列时，我想将我的函数应用到列值上。喜欢
import utilities._
val Logs = sqlContext.read
.format("csv")
.option("header", "true")
.load("dbfs:/mnt/records/Logs/2016.07.17/2016.07.17.{*}.csv")
val Log = Logs.select(
&
浏览 0提问于2018-11-04得票数 0
回答已采纳
2回答
我有一只叫data_clean的熊猫。看起来是这样的：
我想把它转换成一个Spark，所以我使用方法：sparkDF = spark.createDataFrame(data_clean)
但是，这似乎会从原始的dataframe中删除索引列(名为ali、anthony、bill等)。输出.
sparkDF.printSchema()
sparkDF.show()
是
root
|-- transcript: string (nullable = true)
+--------------------+
| transcript|
+----------------
浏览 3提问于2021-07-31得票数 2
回答已采纳
类似于，我想向我的DataFrame添加一个列，它只包含一个空映射。然而，如果我使用该问题的建议答案，则地图的类型是<null,null>，与在那里发布的答案不同。
from pyspark.sql.functions import create_map
spark.range(1).withColumn("test", create_map()).printSchema()
root
|-- test: map(nullable = false)
| |-- key: null
| |-- value: null (valueContainsN
浏览 3提问于2021-12-09得票数 3
回答已采纳
在我的avro模式中，我的十进制列"TOT_AMT“定义为类型”字节“，逻辑类型为”十进制“。
在使用databricks -avro在spark中创建数据框架之后，当我尝试使用sum函数对TOT_AMT列进行求和时，它抛出“函数和需要数值类型而不是Binarytype”错误。
列的定义如下所示，
name="TOT_AMT"，"type":"null"，{“type”：“字节”，“逻辑类型”：“十进制”，“精度”：20，“缩放”：10}
我正在创建数据收集和总结，
val df=sqlContext.read.format("c
浏览 2提问于2017-03-06得票数 1
对于推荐系统，我需要计算整个Spark DataFrame的所有列之间的余弦相似度。
在Pandas中，我经常这样做：
import sklearn.metrics as metrics
import pandas as pd
df= pd.DataFrame(...some dataframe over here :D ...)
metrics.pairwise.cosine_similarity(df.T,df.T)
这会生成列之间的相似性矩阵(因为我使用了转置)
有没有办法在Spark (Python)中做同样的事情？
(我需要将它应用于由数千万行和数千列组成的矩阵，所以这就是为什么我
浏览 3提问于2017-05-12得票数 13
3回答
我在Spark 1.4.0中开始使用Spark SQL和DataFrames。我想用Scala在DataFrames上定义一个自定义的分区程序，但是不知道怎么做。
我正在使用的一个数据表包含一个按帐户划分的事务列表，类似于下面的示例。
Account Date Type Amount
1001 2014-04-01 Purchase 100.00
1001 2014-04-01 Purchase 50.00
1001 2014-04-05 Purchase 70.00
1001 2014-04-01 Paymen
浏览 260提问于2015-06-23得票数 141
回答已采纳
1回答
我有一个返回数据的UDF。就像下面的那个
scala> predict_churn(Vectors.dense(2.0,1.0,0.0,3.0,4.0,4.0,0.0,4.0,5.0,2.0))
res3: org.apache.spark.sql.DataFrame = [noprob: string, yesprob: string, pred: string]
scala> predict_churn(Vectors.dense(2.0,1.0,0.0,3.0,4.0,4.0,0.0,4.0,5.0,2.0)).show
+------------------+------
浏览 3提问于2016-12-20得票数 3
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
