如何从Scala的迭代列表中创建DataFrame？
从Scala的迭代列表中创建DataFrame可以通过以下步骤完成：
- 导入必要的Spark相关库和类：
import org.apache.spark.sql.{SparkSession, Row}
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}- 创建SparkSession对象：
val spark = SparkSession.builder()
.appName("Creating DataFrame from Iteration List")
.master("local")
.getOrCreate()- 定义Schema，描述DataFrame的结构：
val schema = StructType(
Seq(
StructField("name", StringType, nullable = true),
StructField("age", IntegerType, nullable = true)
)
)- 创建一个空的RDD（弹性分布式数据集）：
val rdd = spark.sparkContext.emptyRDD[Row]- 将迭代列表转换为Row对象，并添加到RDD中：
val list = List(("John", 30), ("Lisa", 25), ("Mike", 35))
val rows = list.map(item => Row(item._1, item._2))
val newRdd = rdd.union(spark.sparkContext.parallelize(rows))- 使用SparkSession创建DataFrame：
val df = spark.createDataFrame(newRdd, schema)现在，你就可以使用创建的DataFrame执行各种操作和转换了。
这个答案中没有提及特定的云计算品牌商的产品，因此无需提供相关链接地址。
相关·内容
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
回答已采纳
在Databricks下，我使用此命令从Python Dataframe创建dataframe %python wordsDF = sqlContext.createDataFrame(pandasDF). 我想将数据发送回Azure Datalake Gen2，并且我想使用Scala dataframe。 如何从Pyspark返回/转换Dataframe到scala Dataframe？ ?
浏览 36提问于2021-10-05得票数 0
1回答
我们可以使用以下命令从Java对象列表创建一个dataframe：
DataFrame df = sqlContext.createDataFrame(list, Example.class);
对于Java，Spark可以直接从类推断模式，在本例中是Example.class。
有没有办法在Scala中做同样的事情？
浏览 0提问于2018-04-11得票数 0
1回答
我对scala还很陌生，并且还在努力解决这些问题。如何从dataframe中的颜色中移除列表中的元素，我有一个名称列表，如果它存在于dataframe中，我需要删除它的名称。
我有数据
utid|description
12342|my name is daniel
2345|my name is harry and i love sports
2122|his wife sofia is my schoolmate
和一个列表{“harry”，"daniel"}
输出应该类似于
utid|description
12342|my name is
2345|my name i
浏览 0提问于2018-05-27得票数 0
回答已采纳
1回答
我需要使用SparkSQL从HBase中提取数据，并像SparkSQL一样在数据顶部进行查询。
我所做的事情如下：
创建Spark对象
创建HBase对象
编写JAVPairRDD来获取记录。
我的主要类代码如下：
import java.io.Serializable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
imp
浏览 2提问于2017-01-18得票数 0
我正在学习星火和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
回答已采纳
我的要求是从现有的PySpark程序中调用一个"Spark Scala“函数。将PySpark程序中创建的sparkSession传递给Scala函数的最佳方式是什么？我将我的scala jar传递给Pyspark，如下所示。 spark-submit --jars ScalaExample-0.1.jar pyspark_call_scala_example.py iris.data Scalacode def getDf(spark: SparkSession, query:String, df: DataFrame, log: Logger): DataFrame = {
i
浏览 59提问于2019-10-01得票数 4
1回答
我已经寻找了一段时间，看是否有任何方法可以在Pyspark中使用Scala函数，但我还没有找到关于这个主题的任何文档或指南。
我的目标是使用之前人们定义的scala函数appendPrvlngFields隐式函数。然后我想在python环境中使用这个函数，而不是重新定义它，而是通过一些类型的方式，比如注册scala函数
假设我在Scala中创建了一个使用用户定义的库的简单对象，如下所示：
%scala
package com.Example
import org.library.DataFrameImplicits._
import org.apache.spark.sql.DataFram
浏览 0提问于2020-12-24得票数 0
对于列表中的每个元素，我有一个列表“概要文件”如下所示：
我想从这个列表中创建一个dataframe，这样每个帐户都在dataframe中。当我手动为列表中的每个元素创建一个dataframe，然后将它们全部附加到一个dataframe时，它就能工作了。但是，我想做一个循环，因为我在这个列表中有大约40个元素。
代码如下(手动完成)：
df2 = pd.DataFrame(profiles[1])
df1 = df1.append(df2)
df3 = pd.DataFrame(profiles[2])
df1 = df1.append(df3)
... and so on
我的循环看起来如
浏览 11提问于2022-06-11得票数 0
从一个dataframe中，我想创建一个新的dataframe，其中任何列中至少有一个值为null或空，在Spark1.5/ scala中。
我正在尝试编写一个通用函数来创建这个新的dataframe。在这里，我传递数据文件和列列表，并创建记录。
谢谢
浏览 4提问于2017-07-25得票数 4
回答已采纳
2回答
我正在创建一个火花作业，它要求使用用python编写的函数将列添加到dataframe中。其余的处理是使用Scala完成的。
我已经找到了如何从pyspark调用Java/Scala函数的示例：
我发现用另一种方式发送数据的唯一例子是使用pipe。
我是否可以将整个dataframe发送到python函数，让函数操作数据并添加其他列，然后将结果数据返回给调用的Scala函数？
如果这是不可能的，我目前的解决方案是运行一个pyspark进程，并调用多个Scala函数来操作dataframe，这并不理想。
浏览 3提问于2017-09-15得票数 5
1回答
从列表创建加特林馈送器
、、、
我是gatling和scala的新手。我有一个记录setupValues的列表。我想稍后遍历该列表以测试get Operations。我怎么能做到这一点。 我基本上希望feeder对列表中的给定数据连续重复。 我试着用下面的方式创建feeder。但是它抛出的异常 val setupDeleteExtensions = List.empty[String]
// I populated some values to setupDeleteExtensions List
val extensionIds = Iterator.continually(f
浏览 11提问于2020-04-14得票数 0
回答已采纳
假设df是一个典型的pandas.DataFrame实例，我想了解为什么list(df)会返回一个列名列表。
这里的目标是在源代码中跟踪它，以了解list(<pd.DataFrame>)是如何返回列名列表的。
到目前为止，我发现的最佳资源如下：
摘要:有多种获取DataFrame列名列表的方法，每个方法在性能或惯用约定上各不相同。
摘要: DataFrame遵循一个类似dict的约定，因此使用list()胁迫将返回类似于dict结构的键列表。
：
我在源代码中找不到指向list()如何创建列头名称列表的方法。
浏览 3提问于2022-02-27得票数 1
假设我有以下DataFrame：
scala> val dataset = Seq((0, "hello"), (0, "world"), (0, "hello"), (1, "foo"), (1, "bar")).toDF("id", "text")
dataset: org.apache.spark.sql.DataFrame = [id: int, text: string]
scala> dataset.show()
+---+-----+
| id| text
浏览 1提问于2017-04-12得票数 0
回答已采纳
如何创建具有给定精度的BigDecimal的spark数据集？请参见spark shell中的以下示例。您将看到，我可以创建具有所需BigDecimal精度的DataFrame，但无法将其转换为Dataset。
scala> import scala.collection.JavaConverters._
scala> case class BD(dec: BigDecimal)
scala> val schema = StructType(Seq(StructField("dec", DecimalType(38, 0))))
scala> val h
浏览 73提问于2019-11-14得票数 2
回答已采纳
2回答
我在一个具有相同.txt扩展名的目录中有n个文件，我想在一个循环中加载它们，然后为每个文件创建单独的dataframes。
我读过，但在我的例子中，我的所有文件都有相同的扩展名，我想一个接一个地迭代它们，并为每个文件生成dataframe。
我从计算目录中的文件开始，代码如下
sc.wholeTextFiles("/path/to/dir/*.txt").count()
但是我不知道我应该如何继续下去？请给我引路。
我使用的是Spark 2.3和Scala。
谢谢。
浏览 3提问于2018-08-07得票数 0
回答已采纳
我有以下代码：
fTuple2.cogroup(gTuple2).flatMap { t =>
val fList: ListBuffer[classF] = ListBuffer()
val gList: ListBuffer[classG] = ListBuffer()
while (t._2._2.iterator.hasNext) {
gList.add(t._2._2.iterator.next)
}
val fIter = t._2._1.iterator
while (fIter
浏览 4提问于2015-07-26得票数 0
回答已采纳
2回答
我有一个脚本，它创建一个要连接的数据文件列表。在级联之前，我将检查每个dataframe中的某一列是否存在“1”二进制标志。如果没有，我想从数据文件列表中删除dataframe。我有问题，因为我不知道如何正确索引列表，以删除数据。我用这个代码重新创建了这个问题。
data = {'Name':['Tom', 'Tom', 'Tom', 'Tom'], 'Age':[20, 21, 19, 18]}
data2 = {'Name':['Tom', 'nick&
浏览 1提问于2019-06-27得票数 4
回答已采纳
我正在尝试使用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
回答已采纳
有时(例如，为了测试和标记)，我想强制执行在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
回答已采纳
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
回答已采纳
当我读到一个文件
nums = sc.textfile("hdfs location/input.txt")
我得到了一个字符串列表。
如果在Spark中使用Scala，则可以通过以下方式将数据转换为ints
nums_convert = nums.map(_.toInt)
不过，我不知道如何用火花放电来做同样的事情。我在网上浏览的所有示例都使用脚本本身生成的数字列表，而不是加载文件。或者，文件的格式在dataframe (标签分隔的数据列)中是有意义的。
我尝试使用Python的map，但是它抱怨num是不可迭代的，这是有意义的。最终，我只是尝试处理文件中的数字列表，但不确定
浏览 1提问于2017-09-24得票数 0
回答已采纳
我使用下面的代码随机选择两个元素：
scala.util.Random.shuffle(myList).take(2)
如何确保这两个元素不会被选中两次？
我可以从列表中删除随机元素，创建一个新列表，然后使用与上面相同的代码吗？
浏览 0提问于2014-02-01得票数 1
当创建一个列值默认值为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
回答已采纳
3回答
展示了如何使用Scala case类来推断模式，如何从RDD创建DataFrame。我试图用sqlContext.createDataFrame(RDD, CaseClass)重现这个概念，但是我的DataFrame结果是空的。下面是我的Scala代码：
// sc is the SparkContext, while sqlContext is the SQLContext.
// Define the case class and raw data
case class Dog(name: String)
val data = Array(
Dog("Rex"),
浏览 0提问于2016-05-03得票数 12
回答已采纳
我试图使用星火红移库，并且无法操作由sqlContext.read()命令创建的数据(从redshift读取)。
这是我的代码：
Class.forName("com.amazon.redshift.jdbc41.Driver")
val conf = new SparkConf().setAppName("Spark Application").setMaster("local[2]")
val sc = new SparkContext(conf)
import org.apache.spark.sql._
val sqlContext
浏览 4提问于2015-11-04得票数 1
回答已采纳
最近，我将代码从python转移到scala，并使用scala环境创建了一个项目，其中包括spark，我编写了一个简单的代码来读取sql数据库表，并将其添加为dataframe，并希望用.show()打印前20条记录。
dataframe_mysql = sc.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/new_schema").option("driver", "com.mysql.jdbc.Driver").option("
浏览 0提问于2018-08-29得票数 0
回答已采纳
我有一个来自Scala的DataFrame输出进入其中。我正在寻找从它找回火花DataFrame。 我在databricks中有一个Azure SQL连接，我使用scala进行连接。我可以连接到数据库并输出查询。它为我提供了一个以下Scala格式的Spark DataFrame，我是新手，有人能帮我找回它吗?这样我就可以将它保存为配置单元表 我的scala代码的结果： outputData: Either[org.apache.spark.sql.DataFrame,Boolean] = Left([Product: string, OrderNumber: string ... 27 mo
浏览 13提问于2019-10-17得票数 0
回答已采纳
我是scala的新手，请记住:)
实际上，我有三个问题
如何定义函数以将其传递到df.rdd.mapPartitions，如果我想要创建新行，只需增加几个列
如何在Row对象中添加几个列(或创建一个新列)
如何从已创建的RDD创建DataFrame
先谢谢你
浏览 6提问于2015-11-23得票数 25
回答已采纳
3回答
我写了一个类，它得到一个DataFrame，在它上面做一些计算，并可以导出结果。数据帧由密钥列表生成。我知道我现在正在以一种非常低效的方式做这件事：
var l = List(34, 32, 132, 352) // Scala List
l.foreach{i =>
val data:DataFrame = DataContainer.getDataFrame(i) // get DataFrame
val x = new MyClass(data) // initialize MyClass with new Ob
浏览 1提问于2016-06-28得票数 12
关于如何使用Spark Scala/Python创建空的dataframe/Dataset，有很多示例。但是我想知道如何在Java Spark中创建一个空的dataframe/Dataset。 我必须创建一个空的dataframe，其中只有一列，标题为Column_1，类型为String。
浏览 27提问于2020-07-14得票数 0
回答已采纳
比如说，我有一个org.apache.spark.sql.types.DataType对象的列表，
val tps = [FloatType, LongType, FloatType, DoubleType]，我从dataframe收到的如下所示：
val tps = dataFrame.schema
.filter(f => f.dataType.isInstanceOf[NumericType])
.map(f => f.dataType)
对于这个列表中的每个类型，我需要创建一个具有默认值的对应类型的对象。
[0.0, 0l, 0.0, 0.0]。我怎
浏览 0提问于2019-06-17得票数 0
回答已采纳
enter code here我正在练习在数据仓库中添加一个列表。我可以开发udf并注册，然后在dataframe上应用，但我想尝试一种不同的方法，即提取list from dataframe col和它们map it，然后在新列中提取readd to the original dataframe。
val df = spark.createDataFrame(Seq(("A",1),("B",2),("C",3))).toDF("Str", "Num")
+---+---+
|Str|Num|
+---+---+
浏览 1提问于2022-09-13得票数 0
从Scala列表开始。
val list = List(1,2,3,4)
如何将其转换为TraversableOnce？
浏览 2提问于2013-12-07得票数 6
回答已采纳
我使用Scala从Spark 2.2 dataframe列中提取Ngrams，因此(本例中为trigram)：
val ngram = new NGram().setN(3).setInputCol("incol").setOutputCol("outcol")
如何创建包含所有1到5克的输出列？所以它可能是这样的：
val ngram = new NGram().setN(1:5).setInputCol("incol").setOutputCol("outcol")
但这并不管用。我可以遍历N并为N的每个值创建新的数据帧，但这
浏览 14提问于2018-01-26得票数 3
回答已采纳
1回答
尝试从Windows中的Eclipse程序连接到Windows。
群集是secured using Kerberos身份验证，因此它没有连接到Hbase数据库。
每次我们创建jar文件并在集群中运行时。但是这对于开发和调试并没有用。
如何在类路径中设置hbase-site.xml？
我下载了*site.xml文件，尝试将hbase-site.xml, core-site.xml and hdfs-site.xml作为source文件夹添加，并尝试从项目构建路径中将该文件添加为外部类文件夹，但没有任何工作。我该怎么做呢？
我们是否可以在hbase-site.xml中设置sqlContext，因为
浏览 4提问于2016-11-18得票数 0
1回答
我刚刚开始学习pyspark，这里似乎是一个展示器:我试图将一个本地文本文件加载到spark中：
base_df = sqlContext.read.text("/root/Downloads/SogouQ1.txt")
16/12/29 11:55:20 text.TextRelation:在驱动程序上列出text.TextRelation
base_df.show(10)
16/12/29 11:55:36 INFO storage.MemoryStore:块broadcast_2存储在内存中(估计大小为61.8 KB，空闲78.0 KB) 16/12/29 11
浏览 5提问于2016-12-29得票数 1
我的Databricks笔记本使用的是Python。notebook中的一些代码是用Scala编写的(使用%scala)，其中之一是用于创建数据帧。
如果我再次使用Python/PySpark (默认模式)，我如何使用/访问这个在scala模式下创建的dataframe？
这有可能吗？
谢谢
浏览 14提问于2019-11-17得票数 1
回答已采纳
2回答
我的问题是如何从列表中创建字典列表，是否偏离了问题和问题
我有一个非常长的熊猫数据框架，包括纬度和经度。
LAT LON
40 5
40 6
41 5
42 8
42 9
我成功地将它转换成一个具有towers.values的列表(towers是dataframe的名称)，其中每个列表的格式都是lat，lon。
我的目标是
list_dict = [{'lat': 40, 'lon': 5},
浏览 0提问于2019-09-01得票数 0
回答已采纳
列表可以按如下方式迭代：
scala> val thrill = "Will" :: "fill" :: "until" :: Nil
val thrill: List[String] = List(Will, fill, until)
scala> thrill.map(s => s + "y")
val res14: List[String] = List(Willy, filly, untily)
上面的代码首先创建一个列表，然后第二个命令创建一个包含名为's‘的迭代量的映射，该映射通过附加字符&#
浏览 0提问于2021-05-07得票数 1
我必须解决一个场景，在这个场景中，我使用Scala将HiveContext读取为Future。但是我无法打印返回的dataframe。
我的代码：
val f: Future[DataFrame] = Future{hive.sql("select * from test1.emp")}
如何从Future[DataFrame]对象中获取DataFrame？
浏览 1提问于2017-11-15得票数 0
我正在创建一个函数，它以连接键和条件作为参数，动态地连接两个数据帧。 我理解Spark Scala Dataframe join done the following ways 1) join(right: Dataset[_]): DataFrame
2) join(right: Dataset[_], usingColumn: String): DataFrame
3) join(right: Dataset[_], usingColumns: Seq[String]): DataFrame
4) join(right: Dataset[_], usingColumns: Seq[Str
浏览 93提问于2021-08-18得票数 0
回答已采纳
我试图使用Java从DataFrame中选择列列表。
示例Java代码：
List<String> colList = Arrays.asList(new String[] { "column1", "column2", "column3" });
df.selectExpr((String[])colList.toArray()).show();
在Java中，我必须使用selectExpr而不是select。是否还有使用Java选择列列表的其他方法。
但是在Scala中，我可以做下面这样的事情。
示例Scala代码：
val co
浏览 0提问于2017-12-12得票数 0
我有两个数据帧，第一个数据帧包含实际数据(从CSV文件读取)，第二个数据帧包含一个具有多个关键字的列。即下面的Dataframe-1外观相似(我们必须搜索的地方)：
数据帧-2：
我想要的输出：
我在这里使用spark scala。我想要一个与dataframe-1中的dataframe-2完全匹配的单词。我使用了like、rlike、contains等函数，但它没有给出我想要的输出。有人知道如何在spark scala SQL中或使用dataframe的spark scala函数来开发此逻辑吗？请帮我弄一下这个。
浏览 1提问于2021-02-12得票数 0
我在scala中有一个简单的Spark应用程序。现在，我希望我的spark应用程序只创建一个sparkSession并将Json文件读取到DataFrame中。 object SparkAppExample {
def main(args: Array[String]): Unit = {
val sparkSession = SparkSession.builder()
.appName("Spark Scala Example")
.getOrCreate()
val records: DataFrame = sparkSession.read
浏览 15提问于2020-04-10得票数 0
下面的代码试图从DataFrame中创建SeqT
case class CaseConvert[T: TypeTag](a: T)
def createDf[T: TypeTag](data: Seq[T]): DataFrame = {
spark.createDataFrame(data.map(CaseConvert[T])
}
当上面的createDf方法通过传递类型(例如Seqjava.sql.Timestamp )执行时，它会失败，显示以下错误
UnsupportedOperaionException:不支持TypeTagjava.sql.Timestamp类型的架构
浏览 1提问于2018-02-07得票数 1
具有如下数据帧： df_data=pd.DataFrame({'name':[['ABC','DOS','TRES'],['XYZ','MORTGAGE','SOLUTIONS']],
'original': ['ABC DOS TRES','XYZ MORTGAGE SOLUTIONS']}) 我使用chain.from_iterable来提取列表中的每一项，并将结果添加到数据帧中： s
浏览 65提问于2019-12-18得票数 1
回答已采纳
我正在尝试在Scala/Spark中从现有的DataFrame创建一个新的DataFrame。 下面是我现有的DataFrame： +----------+-----------------------+
| group| value|
+----------+-----------------------+
| 4|[blah blah blah blah...|
| 0|[blah blah blah blah...|
| 1|[blah blah blah blah...|
| 1
浏览 14提问于2020-10-17得票数 1
回答已采纳
我有一个dataframe，它的列在400左右，我想根据我的要求删除100列。因此，我创建了一个Scala列表，其中包含100个列名。然后，我想迭代一个for循环，在每个for循环迭代中实际删除列。
下面是密码。
final val dropList: List[String] = List("Col1","Col2",...."Col100”)
def drpColsfunc(inputDF: DataFrame): DataFrame = {
for (i <- 0 to dropList.length - 1) {
浏览 7提问于2016-09-30得票数 11
回答已采纳
我需要在其中一列中保存包含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
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
