我的CSV文件:
YEAR,UTILITY_ID,UTILITY_NAME,OWNERSHIP,STATE_CODE,AMR_METERING_RESIDENTIAL,AMR_METERING_COMMERCIAL,AMR_METERING_INDUSTRIAL,AMR_METERING_TRANS,AMR_METERING_TOTAL,AMI_METERING_RESIDENTIAL,AMI_METERING_COMMERCIAL,AMI_METERING_INDUSTRIAL,AMI_METERING_TRANS,AMI_METERING_TOTAL,ENERGY_SERVED_RESIDENTIAL,ENERGY_SERVED_COMMERCIAL,ENERGY_SERVED_INDUSTRIAL,ENERGY_SERVED_TRANS,ENERGY_SERVED_TOTAL
2011,34,City of Abbeville - (SC),M,SC,880,14,,,894,,,,,,,,,,
2011,84,A & N Electric Coop,C,MD,135,25,,,160,,,,,,,,,,
2011,84,A & N Electric Coop,C,VA,31893,2107,0,,34000,,,,,,,,,,
2011,97,Adams Electric Coop,C,IL,8334,190,,,8524,,,,,0,,,,,0
2011,108,Adams-Columbia Electric Coop,C,WI,33524,1788,709,,36021,,,,,,,,,,
2011,118,Adams Rural Electric Coop, Inc,C,OH,7457,20,,,7477,,,,,,,,,,
2011,122,Village of Arcade,M,NY,3560,498,100,,4158,,,,,,,,,,
2011,155,Agralite Electric Coop,C,MN,4383,227,315,,4925,,,,,,,,,,
在这里下载Spark代码来读取CSV文件:
public class ReadFile8 {
public static void main(String[] args) throws IOException {
SparkSession session = new SparkSession.Builder().appName("CsvReader").master("local").getOrCreate();
//Data taken by Local System
Dataset<Row> file8Data = session.read().format("com.databricks.spark.csv").option("header", "true").load("file:///home/kumar/Desktop/Eletricaldata/file8_2011.csv");
// Register the DataFrame as a SQL temporary view
file8Data.createOrReplaceTempView("EletricalFile8Data");
file8Data.show();
} …我有一个数据框的列名包含“。” 我想过滤列以获取包含“.”的列名。然后对其进行选择。任何帮助将不胜感激。这是数据集
//dataset
time.1,col.1,col.2,col.3
2015-12-06 12:40:00,2,2,3
2015-12-07 12:41:35,3,3,4
val spark = SparkSession.builder.master("local").appName("my-spark-app").getOrCreate()
val df1 = spark.read.option("inferSchema", "true").option("header", "true").csv("C:/Users/mhattabi/Desktop/dataTestCsvFile/dataTest2.txt")
val columnContainingdots=df1.schema.fieldNames.filter(p=>p.contains('.'))
df1.select(columnContainingdots)
我有一个包含 3 列的数据框，其中一列是整数。我想给它添加一个整数值。
id person age
1 person1 4
2 person4 5
3 person3 7
我想给他们的年龄加上一个常数 20，这样他们的年龄将分别是 24,25 和 27。
我有一个df带有一列的 DataFrame ，column我想转换column成一个向量（例如 a DenseVector），以便我可以在向量和矩阵产品中使用它。
当心：我不需要一列向量；我需要一个矢量对象。
这该怎么做？
我找到了vectorAssembler函数（链接），但这对我没有帮助，因为它将一些 DataFrame 列转换为向量列，这仍然是一个 DataFrame 列；我想要的输出应该是一个向量。
关于这个问题的目标：为什么我要尝试将 DF 列转换为向量？假设我有一个带有数字列的 DF，我需要计算矩阵和该列之间的乘积。我怎样才能做到这一点？（同样适用于 DF 数字行。）欢迎使用任何替代方法。
我需要从 pyspark 数据框中的字符串列中删除正则表达式
df = spark.createDataFrame(
[
("Dog 10H03", "10H03"),
("Cat 09H24 eats rat", "09H24"),
("Mouse 09H45 runs away", "09H45"),
("Mouse 09H45 enters room", "09H45"),
],
["Animal", "Time"],
)
时间戳（例如10H03）是必须删除的正则表达式。
+--------------------+------------------+-----+
| Animal| Animal_strip_time| Time|
+--------------------+------------------+-----+
| Dog 10H03| Dog |10H03|
| Cat 09H24 eats rat| Cat eats rat|09H24|
|Mouse 09H45 runs ...| Mouse runs away|09H45|
|Mouse 09H45 enter...|Mouse enters room|09H45|
+--------------------+------------------+-----+
该列中的时间戳Time可能与该列中的时间戳不同Animal。因此，它不能用于匹配字符串。
正则表达式应遵循 XXHXX 模式，其中 X 是 0-9 之间的数字
如何在一个范围内找到工资？这是我的 Spark SQL 数据框代码，用于查找 10000 到 20000 之间的薪水：
empData.where($"Salary" > 10000 && $"Salary" < 20000).orderBy($"Salary".desc).show()
我相信应该有一个替代解决方案使用between. 我怎样才能使用between方法？
我有一个来自处理部分的数据框，看起来像：
+---------+------+-----------+
|Time |group |value |
+---------+------+-----------+
| 28371| 94| 906|
| 28372| 94| 864|
| 28373| 94| 682|
| 28374| 94| 574|
| 28383| 95| 630|
| 28384| 95| 716|
| 28385| 95| 913|
我想为每个组取（最大时间的值 - 最小时间的值），以获得以下结果：
+------+-----------+
|group | value |
+------+-----------+
| 94| -332|
| 95| 283|
预先感谢您的帮助
我使用的是spark-sql 2.3.1，我设置
spark.sql.shuffle.partitions=40
在我的代码中'
val partitioned_df = vals_df.repartition(col("model_id"),col("fiscal_year"),col("fiscal_quarter"))
当我说
println(" Number of partitions : " + partitioned_df.rdd.getNumPartitions)
它给出 40 作为输出，事实上重新分区后理想情况下计数应该在 400 左右，为什么重新分区在这里不起作用？我在这里做错了什么？如何修复它？
我正在尝试根据其他数据框中的值调整列值之一。这样做时，如果剩余量更多，则需要结转到下一行并计算最终金额。
在此操作期间，我无法保留上一行剩余的金额到下一行操作。我尝试使用滞后窗口功能并采用运行总计选项，但这些选项未按预期工作。
我正在与Scala合作。这是输入数据
val consumption = sc.parallelize(Seq((20180101, 600), (20180201, 900),(20180301, 400),(20180401, 600),(20180501, 1000),(20180601, 1900),(20180701, 500),(20180801, 100),(20180901, 500))).toDF("Month","Usage")
consumption.show()
val consumption = sc.parallelize(Seq((20180101, 600), (20180201, 900),(20180301, 400),(20180401, 600),(20180501, 1000),(20180601, 1900),(20180701, 500),(20180801, 100),(20180901, 500))).toDF("Month","Usage")
consumption.show()
+--------+-----+
| Month|Usage|
+--------+-----+
|20180101| 600|
|20180201| 900|
|20180301| 400|
|20180401| 600|
|20180501| 1000|
|20180601| 1900|
|20180701| 500|
|20180801| 100|
|20180901| 500|
+--------+-----+
val promo = sc.parallelize(Seq((20180101, 1000),(20180201, 100),(20180401, 3000))).toDF("PromoEffectiveMonth","promoAmount")
promo.show()
预期结果：
+-------------------+-----------+
|PromoEffectiveMonth|promoAmount|
+-------------------+-----------+
| 20180101| 1000|
| 20180201| 100|
| 20180401| 3000|
+-------------------+-----------+
val finaldf …我有一个大小为 100GB 的 .csv 文件。我想尽快将它加载到 Spark 中。
Spark 在内部是否将文件分解成块并在多个节点上并行解析块？还是 Spark 仅使用一个节点解析文件并在节点之间分发数据帧？