在pyspark中加入具有相同名称的数据帧
我有两个从两个 csv 文件中读取的数据框. +---+----------+----+|身份证|数量 |RECHARGE_AMOUNT|+---+-----------+-----------------+|1|9090909092|30||2|9090909093|30||3|9090909090|30||4|9090909094|30|+---+-----------+----------
..
apache-spark-2.0相关内容
Spark不会读取第一行中具有空值的列
以下是我的csv文件中的内容: A1，B1，C1A2，B2，C2，D1A3，B3，C3，D2，E1A4，B4，C4，D3A5，B5，C5，E2 因此，有5列，但第一行中只有3个值. 我使用以下命令阅读它: val csvDF:DataFrame = spark.read.option("header"，"false").option(“定界符"，“，").option("infe
..
为运行Spark 2.0的pyspark指定自定义探查器
我想知道如何在PySpark for Spark版本2+中指定自定义探查器类.在1.6以下，我知道我可以这样做: sc = SparkContext('local'，'test'，profiler_cls ='MyProfiler') 但是当我在2.0中创建 SparkSession 时，我没有显式访问权限 SparkContext .有人可以建议如何在Spark 2.0+中执行此操作吗?
..
Spark 2.0 CSV错误
我正在从1.6升级到spark 2，并且在读取CSV文件时遇到问题.在spark 1.6中，我可以将类似的内容读取到CSV文件中. val df = sqlContext.read.format("com.databricks.spark.csv").option("header"，"true").load(文件名) 现在，我使用文档中给出的以下代码: val df = spark.r
..
根据ID列，通过保留顺序将Spark DataFrame分为两个DataFrame(70％和30％)
我有一个火花数据框，就像 id start_time功能1 01-01-2018 3.5671 01-02-2018 4.4541 01-03-2018 6.4552 01-02-2018 343.42 01-08-2018 45.43 02-04-2018 43.563 02-07-2018 34.563 03-07-2018 23.6 我希望能够根据 id列将其分为两个数据帧.因此，我
..
toDF的值不是org.apache.spark.rdd.RDD [(Long，org.apache.spark.ml.linalg.Vector)]的成员
在SPARK 2.0中使用SCALA将LDA之前的转换转换为数据帧时，出现编译错误.引发错误的特定代码如下: val文档= PreLDAmodel.transform(mp_listing_lda_df).select("docId"，“功能").rdd.map {case Row(row_num:Long，features:MLVector)=>(row_num，功能)}.toDF()
..
Scala案例类忽略了Spark shell中的导入
我希望这个问题有明显的答案！ 我刚刚升级到Spark v2.0，并且 spark-shell (Scala 2.11内部版本)有一个奇怪的问题. 如果我输入以下最小Scala， 导入java.sql.Timestamp案例类Crime(caseNumber:字符串，日期:时间戳记，描述:字符串，详细信息:字符串，逮捕:布尔值) 我收到以下错误消息， :11:错
..
Cassandra的Spark2会话，SQL查询
在Spark-2.0中，什么是创建Spark会话的最佳方法.因为在Spark-2.0和Cassandra-中都对API进行了重新设计，从而实质上弃用了SqlContext(以及CassandraSqlContext).因此，为了执行SQL，我要么创建一个Cassandra会话(com.datastax.driver.core.Session)并使用execute(“").或者我必须创建一个 Spa
..
任务仅在Spark中的一个执行程序上运行
我正在使用Java在spark代码下运行. 代码 Test.java package com.sample;导入org.apache.spark.SparkConf;导入org.apache.spark.SparkContext;导入org.apache.spark.sql.Dataset;导入org.apache.spark.sql.Row;导入org.apache.spark.
..
Spark DataFrame:查找并设置子级的主根
我有以下Apache Spark数据框: 父母-孩子 A1 - A10 A1 - A2 A2 - A3 A3 - A4 A5 - A7 A7 - A6 A8 - A9 此DataFrame显示父级和子级之间的连接.逻辑上看起来像这样: 输出 + ------ + ----- +|父母|孩子|+ ------ + ----- +|A1 |A10 ||A1 |A11
..
从包含Option [T]的行创建DataFrame的问题
我正在将一些代码从Spark 1.6迁移到Spark 2.1，并遇到以下问题: 这在Spark 1.6中效果很好 导入org.apache.spark.sql.types.{LongType，StructField，StructType}val模式= StructType(Seq(StructField("i"，LongType，nullable = true)))val行= sparkC
..
为什么SparkSQL在SQL查询中需要两个文字转义反斜杠?
当我从Spark 2.0 REPL(spark-shell)运行以下Scala代码时，它将按我的预期运行，并使用简单的正则表达式拆分字符串. 导入org.apache.spark.sql.SparkSession//创建会话val sparkSession = SparkSession.builder.master("local").getOrCreate()//使用SparkSQL分割字符串v
..
java.lang.IllegalStateException:读取增量文件时出错，使用kafka进行火花结构化流式传输
我在我们的项目中使用结构化流+ Kafka进行实时数据分析.我正在使用Spark 2.2，kafka 0.10.2. 在应用程序启动时从检查点进行流查询恢复时，我遇到了一个问题.由于从单个kafka流点派生了多个流查询，并且每个流查询都有不同的checkpint目录.因此，在作业失败的情况下，当我们重新启动作业时，会有一些流查询无法从检查点位置恢复，因此会引发读取增量文件错误的异常.这是日志
..
在Scala Spark2中将saveAsNewAPIHadoopDataset运行到HBase时获取空指针异常
我正在使用saveAsNewAPIHadoopDataset将puts RDD保存到Hbase.以下是我的职位创建和提交. val outputTableName = "test3" val conf2 = HBaseConfiguration.create() conf2.set("hbase.zookeeper.quorum", "xx.xx.xx.xx")
..
Spark流的动态分配
我有一个Spark Streaming作业与其他作业（Spark核心作业）一起在我们的集群上运行。我想对包括Spark Streaming在内的这些作业使用动态资源分配。根据下面的JIRA问题，动态分配不支持Spark Streaming（在1.6.1版本中）。但是已在2.0.0中修复 JIRA链接 PDF在本期中，它说应该有一个名为 spark.streaming.dynamicA
..
使用Spark 2.0.2从Kafka读取Avro消息(结构化流)
我有一个spark 2.0应用程序，该应用程序使用Spark Streaming(带有spark-streaming-kafka-0-10_2.11)从kafka读取消息. 结构化的流媒体看起来真的很酷，所以我想尝试迁移代码，但是我不知道如何使用它. 在常规流中，我使用kafkaUtils来创建Dstrean，在传递的参数中，它是值反序列化器. 在结构化流式传输中，文档说我应该使
..
如何从单个目录仅加载pyspark spark.read.csv中的前n个文件
在一种情况下，我正在加载和处理4TB数据， 大约一个文件夹中有15000个.csv文件. 由于我的资源有限，因此我计划将这些资源分成两部分进行处理 批次并将它们合并. 我试图了解我是否只能加载50％(或前n个 批处理1中的文件数，批处理2中的其余文件)使用 spark.read.csv. 我无法使用正则表达式，因为生成了这些文件 来自多个来源，数量不均(某些来源 来源很少，而其他来
..
在Spark DataFrame中按列运行创建唯一的分组键
我与此类似，其中spark是我的sparkContext.我已经在sparkContext中导入了implicits._，因此可以使用$语法: val df = spark.createDataFrame(Seq(("a", 0L), ("b", 1L), ("c", 1L), ("d", 1L), ("e", 0L), ("f", 1L))) .toDF("id
..
根据Spark中的列值拆分数据集
我正在尝试根据“制造商"列的内容将数据集拆分为不同的数据集.这很慢 请提出一种改进代码的方法，以便它可以更快地执行并减少Java代码的使用. List lsts= countsByAge.collectAsList(); for(Row lst:lsts){ String man=lst.toString();
..
使用extraOptimizations转换Spark SQL AST
我想将一个SQL字符串作为用户输入，然后在执行之前对其进行转换.特别是，我想修改顶层投影(选择子句)，注入要由查询检索的其他列. 我希望通过使用sparkSession.experimental.extraOptimizations连接到Catalyst来实现这一目标.我知道我要尝试的并不是严格意义上的优化(转换会更改SQL语句的语义)，但是API似乎仍然合适.但是，查询执行程序似乎忽略了我
..