如何在Scala dataframe中获取列的数据类型
在Scala dataframe中获取列的数据类型可以使用dtypes方法。该方法返回一个包含列名和数据类型的元组数组。每个元组包含两个元素，第一个元素是列名，第二个元素是数据类型。
以下是获取列数据类型的示例代码：
import org.apache.spark.sql.{SparkSession, DataFrame}
val spark = SparkSession.builder()
.appName("Get column data types")
.master("local")
.getOrCreate()
val data = Seq(
(1, "John", 25),
(2, "Jane", 30),
(3, "Bob", 35)
)
val df: DataFrame = spark.createDataFrame(data).toDF("id", "name", "age")
val columnTypes = df.dtypes
columnTypes.foreach { case (columnName, dataType) =>
println(s"Column $columnName has data type $dataType")
}输出结果为：
Column id has data type IntegerType
Column name has data type StringType
Column age has data type IntegerType在这个例子中，我们首先创建了一个SparkSession对象，然后使用createDataFrame方法创建了一个DataFrame对象。接下来，我们使用toDF方法为每列指定了列名。最后，我们使用dtypes方法获取了列的数据类型，并通过遍历输出了每列的名称和数据类型。
推荐的腾讯云相关产品：腾讯云分析型数据库 TDSQL、腾讯云数据仓库 ClickHouse、腾讯云弹性MapReduce TEMR。
腾讯云产品介绍链接地址：
相关·内容
当创建一个列值默认值为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回答
我在Databricks上使用Scala，并且：
I有一个有N个列的dataframe。除了第一个Y列，所有的Y列都是"float“类型，并且我想把数字舍入到0小数位。
我不想为每一列写需要四舍五入的特定代码行，因为可能有很多列需要四舍五入，而且它们各不相同。
为了做到这一点，我尝试用Map创建一个函数(不确定它是否是最好的选项)：
def roundValues(precision: Int)(df: DataFrame): DataFrame = {
val roundedCols = df.columns.map(c => round(col(c), precision)
浏览 7提问于2022-02-19得票数 0
回答已采纳
我有一个简单的spark作业，它从文件中拆分单词并加载到hive中的表中。
public static void wordCountJava7() {
// Define a configuration to use to interact with Spark
SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("Work Count App");
SparkContext sc = new SparkContext(conf);
// Crea
浏览 1提问于2016-04-29得票数 1
我正在尝试执行星火的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
回答已采纳
2回答
我在Scala中有一个列名列表，如下所示
var cols = List("col1", "col2", "col3","col4")
另外，我有一个包含这些列的数据框架，但都是字符串。现在，我想通过迭代dataframe的列表或列来转换dataframe的列，因为我的列列表非常大，而且我不能使用这么多的.withColumn参数。
提前谢谢
浏览 5提问于2016-01-28得票数 3
回答已采纳
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
回答已采纳
我正在寻找一种动态处理数据类型转换的方法。SparkDataframes，我正在使用hive SQL将数据加载到Dataframe中，并将其存储到dataframe中，然后写入到parquet文件中。Hive无法读取某些数据类型，我希望将decimal数据类型转换为Double。是否有任何方法可以动态处理数据类型，而不是单独指定每个列的名称。假设在我的dataframe中，8列中有50列是小数，并且需要将所有8列转换为双精度数据类型，而不指定列名。我们可以直接这么做吗？
浏览 3提问于2017-05-16得票数 0
日安。我正在部署一个流作业，以将数据从Spark (Scala)插入到Postgres。
df.select("col1","col2").write.mode(SaveMode.Append).jdbc(url, "tableName", connectionProperties)
这里，col2在dataframe中有uuid值，但它是一个字符串数据类型。当它试图插入到将col2列定义为uuid类型的表时，它在Column is of type uuid but expression is of type character varying中
浏览 3提问于2021-02-25得票数 0
回答已采纳
我需要在我的DataFrame中生成一个带有随机时间戳的新列，它有几秒钟的时间。DataFrame包含10.000行。起始时间戳应该是1516364153。我试图解决以下问题：
df.withColumn("timestamp",lit(1516364153 + scala.util.Random.nextInt(2000)))
但是，所有时间戳都等于某些特定值，例如，1516364282，而不是许多不同的值。可能有一些重复，但是为什么所有的值都是相同的呢？看起来只生成了一个随机数，然后在整个列中传播。
我该如何解决这个问题？
浏览 1提问于2018-02-10得票数 0
回答已采纳
3回答
输入数据可能并不总是所有的列。在SQL或SCALA中，我希望创建一个select语句，即使dataframe没有列，它也不会出错，它只会输出存在的列。
例如，这个语句可以工作。
Select store, prod, distance from table
+-----+------+--------+
|store|prod |distance|
+-----+------+--------+
|51 |42 |2 |
|51 |42 |5 |
|89 |44 |9 |
如果dataframe如下所示，我希望相同的语句工作，只需忽略不
浏览 0提问于2019-03-18得票数 2
回答已采纳
我想从dataframe访问特定列的min和max，但是我没有该列的标题，只有它的编号，所以我应该使用scala吗？
也许是这样的：
val q = nextInt(ncol) //we pick a random value for a column number
col = df(q)
val minimum = col.min()
对不起，如果这个问题听起来很傻，但我找不到关于这个问题的任何信息:/
浏览 10提问于2017-04-05得票数 35
回答已采纳
我知道使用.withColumn()向星火.withColumn()添加新列的方法，以及返回DataFrame的UDF。我还知道，我们可以将结果DataFrame转换为DataSet。
我的问题是：
如果我们仍然遵循传统的DF方法(即将列名作为UDF输入的字符串传递)，DataSet的类型安全性是如何发挥作用的？
是否有一种“面向对象的方式”来访问列(而不是将列名作为字符串传递)，就像我们以前使用RDD那样，用于追加一个新列。
如何在地图、过滤器等正常操作中访问新列？
例如：
scala> case class Temp(a : Int, b : String
浏览 1提问于2016-11-15得票数 11
回答已采纳
我正在学习星火和Scala，并在星火REPL中进行实验。
当我试图将列表转换为DataFrame时，它的工作方式如下：
val convertedDf = Seq(1,2,3,4).toDF("Field1")
但是，当我尝试将列表列表转换为具有两列( DataFrame，field1，field2)的列表时，
java.lang.IllegalArgumentException:需求失败:列数不匹配
错误信息：
val twoColumnDf =Seq(Seq(1,2,3,4,5), Seq(5,4,3,2,3)).toDF("Field1", (Fi
浏览 1提问于2018-06-22得票数 0
回答已采纳
1回答
我有个功能
def f(tuples: Seq[Product]) : Unit = if (tuples.size>0) {
// etc using tuples.foreach( x=> { ... x.productIterator.foreach(...) })
}
需要将其称为传递一个DataFrame，f( myDF.rdd.? )。如何通过dataframe调用函数？
(或者怎么投？如何实现collectAsTuples?)
PS:需要一个“通用强制转换”，可能是rdd.map(genericTuple)，而不是一个解决方案专门化的元组。
请注意:他们被
浏览 2提问于2019-11-21得票数 0
有时(例如，为了测试和标记)，我想强制执行在DataFrame上定义的转换。AFAIK调用像count这样的操作并不能确保所有的Columns都是实际计算的，show可能只计算所有Rows的一个子集(参见下面的示例) 我的解决方案是使用df.write.saveAsTable将DataFrame写到HDFS，但是这会“扰乱”我的系统，我不想再保存更多的表。 那么触发DataFrame求值的最佳方式是什么呢 编辑： 请注意，在spark开发人员列表上还有一个最近的讨论：http://apache-spark-developers-list.1001551.n3.nabble.com/Will-
浏览 62提问于2017-03-10得票数 20
回答已采纳
我正在寻找Scala中的一个命令，以在DataFrame中的所有列中查找空值之和。
在python中，我可以使用以下命令获得它：Df.isnull().sum()
你能告诉我同样的scala命令吗？
浏览 4提问于2022-10-22得票数 0
我试图将json文件添加到JSON文件中，以便访问JSONSerDe数据，将JSON从spark作业加载到hive表中。我的代码如下所示：
SparkConf sparkConf = new SparkConf().setAppName("KafkaStreamToHbase");
JavaSparkContext sc = new JavaSparkContext(sparkConf);
JavaStreamingContext jssc = new JavaStreamingContext(sc, Durations.seconds(10)
浏览 4提问于2016-06-14得票数 2
我在目标表中有一个固定的火花DataFrame订单：
目标(col1字符串、col2 int、col3字符串、col4 double)
现在，如果源数据以混乱的顺序出现：
源(col3字符串、col2 int、col4 double、col1字符串)。
如何使用DataFrame重新排列源DataFrame以匹配目标DataFrame的列顺序？
源应该重新排序，如下所示，以匹配目标DataFrame：
输出：
更新的源(col1字符串、col2 int、col3字符串、col4 double)
设想2：
源Dataframe =a，c，d，e
目标数据=a，b，c，d
浏览 0提问于2019-06-27得票数 0
回答已采纳
1回答
SSIS数据转换
、、、、
在SSIS中，我读取列格式的csv文件，例如1.25或2.50。在数据转换任务中，我将转换为十进制dt_decimal scala 2。在数据表中，列的格式为十进制(18，2)。
数据将用125.00或25.00来存储，而不是1.25和2.50。
我要调整什么？
浏览 0提问于2019-02-16得票数 2
回答已采纳
1回答
非常慢的地板读
、、
我正在尝试从两个不同的位置读取拼图文件A和B。它们都是GCP中的parquet文件，模式中的列数大致相同(80-90列，大部分是字符串)。B在文件大小和记录计数方面非常小(大约比A小5个数量级)。但是我想知道为什么会是这样，从GS那里读到的东西和A几乎一样。
scala> show_timing{spark.read.parquet("gs://bucket-name/tables/A/year=2018/month=4/day=5/*")}
Time elapsed: 34862525 microsecs
res5: org.apache.spark.sql.DataF
浏览 0提问于2018-04-13得票数 2
我有一个包含布尔列(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
我刚开始使用SparkSQL/Scala，我正在为一些看似简单的任务而奋斗。
我正在尝试从Scala字符串数组构建一些动态SQL。我正在尝试在我的DataFrame中重新键入一些列，但是在运行时，在DataFrame中可以看到列集的地方，我才会确切地知道需要重新键入哪些列。所以我试着这么做：
val cols = df.columns
val typedCols = cols.map( c => getTypedColumn(c) )
df.select( ...) or df.selectExpr(...) // how to invoke this with vals from m
浏览 1提问于2017-12-07得票数 1
回答已采纳
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
回答已采纳
我正在尝试在C# Spark中实现向量自定义函数。 我已经通过Spark .Net创建了.Net Spark环境。在我的IntegerType专栏中，Vector Udf (Apache箭头和Microsoft.Data.Analysis都是)很好用。现在，尝试将Integer数组类型的列发送到Vector Udf，但找不到实现此目的的方法。 用法 using System;
using System.Linq;
using Microsoft.Data.Analysis;
using Microsoft.Spark.Sql;
using func = Microsoft.Spark.Sql
浏览 17提问于2021-03-25得票数 6
我有一个json字符串：
{"user_rating": {"rating_text": "Excellent", "rating_color": "3F7E00", "votes": "778", "aggregate_rating": "4.5"}}
我希望用DataFrame列创建一个DataFrame，如下所示：
rating_text | rating_color | votes | aggregate_rating
当我将其编码为：
val
浏览 5提问于2020-01-19得票数 2
回答已采纳
我有一个包含日期时间列的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
回答已采纳
我在dataframe列中有一个s3路径，我需要该路径的目录大小。在spark scala中有没有什么方法可以让我们在一个新的列中获得s3目录的大小。
浏览 24提问于2021-05-10得票数 0
回答已采纳
我有一个包含float和double值的dataframe。
scala> val df = List((Float.NaN, Double.NaN), (1f, 0d)).toDF("x", "y")
df: org.apache.spark.sql.DataFrame = [x: float, y: double]
scala> df.show
+---+---+
| x| y|
+---+---+
|NaN|NaN|
|1.0|0.0|
+---+---+
scala> df.printSchema
root
|-- x: fl
浏览 2提问于2017-05-12得票数 3
回答已采纳
我正在尝试使用dataframe中的现有列向dataframe中添加更多的列。然而，Scala是不可变的，因此很难进行迭代。因此，我想出了一个for循环来输出字符串(参见下面的示例代码，它存储了我可以使用的整个语句)。
val train_df = sqlContext.sql("select * from someTable")
/*for loop output is similar to the Str variable as below*/
var Str = ".withColumn(\"newCol1\",$\"col1\"
浏览 0提问于2018-08-15得票数 0
回答已采纳
在Spark中，将Timestamp列转换为毫秒时间戳Long列的最短和最有效的方法是什么？
下面是一个从时间戳到毫秒的转换示例
scala> val ts = spark.sql("SELECT now() as ts")
ts: org.apache.spark.sql.DataFrame = [ts: timestamp]
scala> ts.show(false)
+-----------------------+
|ts
浏览 1提问于2019-06-18得票数 2
2回答
在我的hadoop集群上执行spark2-submit时，当在hdfs中读取.jsons目录时，我不知道如何解决它。
我在几个黑板上发现了一些关于这方面的问题，但没有一个是很受欢迎的，也没有一个有答案。
我尝试过显式导入org.apache.spark.sql.execution.datasources.json.JsonFileFormat，但导入SparkSession似乎是多余的，因此没有得到认可。
不过，我可以确认这两个类都是可用的。
val json:org.apache.spark.sql.execution.datasources.json.JsonDataSource
val
浏览 0提问于2020-07-05得票数 0
回答已采纳
Spark版本2.4.0集群: AWS EMR
我正在将tsv.gz输入文件读入到spark dataframe中，当尝试将其存储回来时，无论是否进行一些转换，如果推断模式为真，则会得到如下所示的异常。
tsv_data = spark. \
read. \
format("csv"). \
option("delimiter", "\t"). \
option("header", "true"). \
option("inferSchema", "t
浏览 3提问于2019-09-04得票数 1
我有两个具有map数据类型列的dataframe。我尝试使用传统的except方法来获取两个数据帧之间的差异，但是我得到了下面的错误。
scala> val outputDF = Seq(
| (1, "Visa", 0, Map("Visa" -> 1)),
| (2, "MC", 2, Map("Visa" -> 1, "MC" -> 1)),
| (3, "Amex", 0, Map("Amex" -> 1)),
浏览 1提问于2020-10-28得票数 0
我能够创建数据帧并通过以下方式强制使用一种数据类型
import pandas as pd
test = pd.DataFrame({'a':[1,2,3], 'b':[1.1,2.1,3.1]}, dtype=int)
但是我想为每一列指定类型。我该怎么做呢？我尝试了以下方法，但不起作用，因为结果数据类型是对象，并且b列不会强制转换为整数。
test = pd.DataFrame({'a':[1,2,3], 'b':[1.1,2.1,3.1]}, dtype=[('a', int),('b', i
浏览 0提问于2014-03-26得票数 10
如果我有一个包含ArrayString列的Dataframe
scala> y.show
+---+----------+
|uid|event_comb|
+---+----------+
| c| [xx, zz]|
| b| [xx, xx]|
| b| [xx, yy]|
| b| [xx, zz]|
| b| [xx, yy]|
| b| [xx, zz]|
| b| [yy, zz]|
| a| [xx, yy]|
+---+----------+
如何将列"event_comb"分成两列(例如"event1"和
浏览 1提问于2018-02-09得票数 6
回答已采纳
我有一个情绪分析程序，可以使用递归神经网络来预测给定的电影评论是正面的还是负面的。我正在使用Deeplearning4j深度学习库来实现这个程序。现在，我需要将该程序添加到apache spark管道中。
这样做的时候，我有一个扩展org.apache.spark.ml.classification.ProbabilisticClassifier的类MovieReviewClassifier，我必须在管道中添加该类的一个实例。使用setFeaturesCol(String s)方法将建立模型所需的特征输入程序。我添加的功能是String格式的，因为它们是一组用于情感分析的字符串。但是这些特性应
浏览 3提问于2016-02-19得票数 1
2回答
我有一张桌子 DEST_COUNTRY_NAME ORIGIN_COUNTRY_NAME count
United States Romania 15
United States Croatia 1
United States Ireland 344 我把上面的代码转换成了DataFrame val flightData2015 = spark
.read
.option("inferSchema", "true")//infers the input schema automatically from data
.option("he
浏览 19提问于2019-02-08得票数 0
回答已采纳
我是Spark和Scala的新手。我试图调用一个函数作为Spark UDF，但我遇到了这个错误，我似乎无法解决。
我知道在Scala中，Array和Seq是不同的。WrappedArray是Seq的一个子类型，并且在WrappedArray和数组之间存在隐式转换，但我不确定为什么在使用自定义函数的情况下不会发生这种转换。
任何能帮助我理解和解决这个问题的建议都是非常感谢的。
这是一段代码片段
def filterMapKeysWithSet(m: Map[Int, Int], a: Array[Int]): Map[Int, Int] = {
val seqToArray = a.toArra
浏览 0提问于2016-10-23得票数 23
回答已采纳
当我试图查看从拼图文件创建的dataframe中的数据时，我遇到了下面的错误。 Expected: decimal(16,2), Found: BINARY
at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:221)
at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:130)
浏览 153提问于2020-04-30得票数 2
Scala2.10在这里使用Spark1.6.2。我有一个与类似的(但不是相同的)问题，然而，被接受的答案不是，它假定了关于火花的某种“预先知识”；因此我无法复制它或理解它。(更重要的是，)，这个问题也仅限于向现有的dataframe中添加一个新列，而我需要添加一个列以及数据中所有现有行的值。
因此，我希望向现有的DataFrame添加一列，然后将该新列的初始(“默认值”)值应用到所有行。
val json : String = """{ "x": true, "y": "not true" }""
浏览 4提问于2016-10-10得票数 7
回答已采纳
以下代码将csv读取到scala中的dataframe中：
val mDF: DataFrame = spark.read.csv("src/test/resources/knimeMerged.csv")
但是，它将导入数据的第一行视为数据行。实际上，第一行是headers。它使用数据帧的默认标头作为标头(例如，_c0、_c1)
我假设有一个选项允许导入csv文件的头文件，但在Scala API文档中找不到它(我是scala及其文档的新手)。
任何关于选项是什么以及如何实现的提示都将不胜感激
浏览 12提问于2017-08-31得票数 0
回答已采纳
我正在尝试将一些计算列添加到SparkR数据框中，如下所示：
Orders <- withColumn(Orders, "Ready.minus.In.mins",
(unix_timestamp(Orders$ReadyTime) - unix_timestamp(Orders$InTime)) / 60)
Orders <- withColumn(Orders, "Out.minus.In.mins",
(unix_timestamp(Orders$OutTime) - unix_timestamp(Orders$InTime)) / 6
浏览 2提问于2015-12-29得票数 0
我需要用脚本编写一个DataFrame到Azure事件中心。
我的解决方案基于本文。
如何将具有多列的Dataframe转换为具有名为“body”的单一列的DataFrame，并将所有列作为json？
我正在使用Scala和Spark 3
下面是我找到的一个解决方案，toJson在其中创建了一个名为“value”的列。
df.toJSON.selectExpr("value as body")
谢谢。
浏览 8提问于2022-01-11得票数 0
我使用的是星星之火-sql.2.4.1v，datastax-java-cassandra-connector_2.11-2.4.1.jar和java8。
我创建cassandra表如下所示：
create company(company_id int PRIMARY_KEY, company_name text);
JavaBean如下所示：
class CompanyRecord(
Integer company_id;
String company_name;
//getter and setters
//default & parametarized constructors
浏览 2提问于2019-10-28得票数 1
我有XML文件，其中包含使用withColumn函数在DataFrame上运行的所有转换，如下所示:如何在DataFrame上应用它。
我有一个使用Scala ToolBox和runTmirror编写的代码，它在内部编译代码并在DataFrame上运行这些规则。它能很好地工作在不到100列的地方。但是现在需求已经改变了，列的数量从80列增加到210列，所以这段代码失败了，应该是StackOverflow error。这是Scala2.11 ()的未决问题。
因此，我想使用任何星火工具，而不是Scala ToolBox。我也尝试过使用foldLeft，但由于无法传递列函数(如lit或concat
浏览 0提问于2019-08-31得票数 0
回答已采纳
我有一个非常大的星火DataFrame和许多列，我想作出一个明智的判断是否保持在我的管道中，部分取决于它们有多大。所谓“有多大”，我指的是缓存这个DataFrame时内存中字节的大小，我希望这是对处理这些数据的计算成本的一个不错的估计。有些列是简单类型(例如，双列、整数列)，而另一些列是复杂类型(例如数组和可变长度映射)。
我尝试过的一种方法是在没有列的情况下缓存DataFrame，然后查看Spark中的Storage，然后取不同的内容。但是对于一个列很多的DataFrame来说，这是一个恼人而缓慢的练习。
我通常使用PySpark，所以最好使用PySpark回答，但是Scala也可以。
浏览 0提问于2019-02-25得票数 2
回答已采纳
1回答
我得把火种翻译成scala。我有一个像这样的函数：lambda x: ';'.join(map(str, x)。然后我们用
our_dataframe.groupBy('column').agg(our_lambda_function(F.collect_set('other_col')))
因此，我们首先按某个列进行分组，然后作为字符串数组获取值。我们的lambda函数从这个数组中生成一个字符串。
举例说明。而不是
\x列
\x{e76f}a，b，c，d，e，f
我们想要
\x列
A；b；c；d；e；f
如何用scala做这个呢？
浏览 2提问于2021-08-29得票数 0
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
回答已采纳
我试图在Scala中创建一个函数来过滤最近的日期。我想保持它的一般性，因此，无论我输入的数据，只要它有列“日期”，将返回我的最新行的数据。值得注意的是，我的date列通常被定义为一个字符串，格式为yyyy。我很确定这里的代码是有缺陷的，但我想它说明了这个想法。
def fixDate(table: DataFrame): DataFrame = {
table
.withColumn("date", from_unixtime(unix_timestamp(col("date"), "yyyy-MM-dd"), "
浏览 3提问于2022-08-07得票数 0
8回答
我们正在读取来自MongoDB Collection的数据。Collection列有两个不同的值(例如：(bson.Int64,int) (int,float) )。
我正在尝试使用pyspark获取一个数据类型。
我的问题是有些列有不同的数据类型。
假设quantity和weight是列
quantity weight
--------- --------
12300 656
123566000000 789.6767
1238 56.22
345 23
34
浏览 42提问于2017-07-11得票数 62
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
