spark 为每组动态创建 struct/json
我有一个像火花数据框 +-----+---+---+---+------+|组|一个|乙|c|配置|+-----+---+---+---+------+|一个|1|2|3|[a]||乙|2|3|4|[a, b]|+-----+---+---+---+------+val df = Seq(("a", 1, 2, 3, Seq("a")),("b", 2, 3,4, Seq("a", "b")))
..
apache-spark-sql相关内容
如何使用 saveAsHadoopFile 或 MultiTextOutputFormat 保存数据框
基本上我需要根据 DataPartition 列创建输出文件.数据框中的最后一列 所以第一行和最后一行将保存在Fundamental.Fundamental.Fundamental.Japan.1.2018-09-24-0937.Full.txt中间一行将保存在Fundamental.Fundamental.Fundamental.ThirdParty.1.2018-09-24-0937.F
..
Spark Sql - 插入外部 Hive 表错误
我正在尝试通过 spark sql 将数据插入到外部配置单元表中.我的蜂巢表是通过一列存储的.创建外部配置单元表的查询是这样的 create external table tab1 (col1 type,col2 type,col3 type) 由 (col1,col2) 聚类并由 (col1) 排序成 8 个存储为 parquet 的桶 现在我尝试将 Parquet 文件(存储在 hdfs
..
尝试覆盖 Hive 分区时写入 __HIVE_DEFAULT_PARTITION__ 的损坏行
在尝试使用 Spark 2.3 覆盖 Hive 表中的分区时，我看到了一些非常奇怪的行为 首先，我在构建 SparkSession 时设置以下设置: .config("spark.sql.sources.partitionOverwriteMode", "dynamic") 然后我将一些数据复制到新表中并按 date_id 列进行分区. ds.写.format("镶木地板").opti
..
如何基于相等性检查在 Spark 中使用内部数组查询嵌套 json
我有一个嵌套的 json 结构加载到 spark 中的数据帧中.它包含多层数组，我正试图弄清楚如何通过内部数组中的值来查询这个结构. 示例:考虑以下结构(directors.json 文件) [{“导演":“史蒂文·斯皮尔伯格"，“电影":[{"name": "E.T",“演员":[“亨利·托马斯"、“德鲁·巴里摩尔"]},{"name": "The Goonies",“演员":[“肖恩·
..
Spark 从多列 DataFrame 批量写入 Kafka 主题
批处理后，Spark ETL 我需要将包含多个不同列的结果数据帧写入 Kafka 主题. 根据以下 Spark 文档 https:///spark.apache.org/docs/2.2.0/structured-streaming-kafka-integration.html 写入 Kafka 的 Dataframe 应该在架构中包含以下强制性列: 值(必需)字符串或二进制 正
..
Pyspark 数据帧中的 Timedelta - TypeError
我正在使用 pyspark 2.3.1 开发 Spark 2.3、Python 3.6 我有一个 Spark DataFrame，其中每个条目都是一个工作步骤，我想将一些行合并到一个工作会话中.这应该在下面的函数 getSessions 中完成.我相信它有效. 我进一步创建了一个包含我想要的所有信息的 RDD - 每个条目都是一个带有所需列的 Row 对象，看起来类型很好(一些数据被伪
..
如何在 Scala 中使用 spark cassandra 连接器 API
我之前的帖子:修复准备好的 stmt 警告. 我无法解决它，建议很少，我尝试使用 spark cassandra 连接器来解决我的问题.但我对它在我的应用程序中的使用感到完全困惑.我尝试编写如下代码，但不确定如何使用 API. val conf = new SparkConf(true).set("spark.cassandra.connection.host", "1.1.1.1").s
..
使用 Scala 以优化的方式将 Spark 数据帧的 Not null 值替换为“1"
我有一个名为 freq 的输入 spark-dataframe as， +---------------+----+----+----+----+|Main_CustomerID|A1|A2|A3|A4|+--------------+----+----+----+----+|101|空|2|1|空||102|2|空|2|4||103|1|2|空|3||104|2|空|3|空|+--------
..
在 SparkSQL 中使用窗口函数 (dense_rank()) 进行选择
我有一个包含客户购买记录的表格，我需要指定购买是在特定日期时间窗口内进行的，一个窗口是 8 天，所以如果我今天购买了 5 天内购买了一次，那么如果窗口号是我的购买1，但如果我在今天的第一天和 8 天后的第二天这样做，第一次购买将在窗口 1 中，最后一次购买将在窗口 2 中 创建临时表事务(client_id int,transaction_ts 日期时间，store_id int)插入交易值(1
..
如何强制 spark 避免数据集重新计算?
我有一个从 cassandra 中加载的数据集.加载此数据集后，我将从 cassandra 中删除一些项目，但我希望我的数据集作为下一个计算的第一个.我已经使用 persist(DISK_ONLY) 来解决它，但它似乎是尽力而为.如何强制 spark 避免重新计算? 示例: val dataset:Dataset[Int] = ???//来自 cassandra 的东西dataset.p
..
如何制作整数索引行?
我有一个数据帧: +-----+--------+---------+|usn|log_type|item_code|+-----+--------+---------+|0|11|I0938||916|19|I0009||916|51|I1097||916|19|C0723||916|19|I0010||916|19|I0010||12331|19|C0117||12331|19|C0117|
..
为什么 HiveConnection 不支持 getHoldability()?将 C3P0 与 HiveDriver 一起使用
我正在尝试将 C3P0 (com.mchange.v2.c3p0.ComboPooledDataSource) 与 HiveDriver (org.apache.hive.jdbc.HiveDriver).我有一个例外: java.sql.SQLException:不支持方法在 org.apache.hive.jdbc.HiveConnection.getHoldability(HiveConn
..
根据带条件的时间戳获取数据框中的最新记录
我的问题标题可能不准确，但我希望我能够解释我的问题所以我有一个像下面这样的数据框 DataPartition_1|^|PartitionYear_1|^|TimeStamp|^|OrganizationId|^|AnnualPeriodId|^|InterimPeriodId|^|InterimNumber_1|^|FFAction_1SelfSourcedPublic|^|2001|^|151
..
SparkR - 转换为日期格式
如何使用 Spark 数据帧的特定格式将字符串转换为日期? 在 dplyr 中，我会这样做: df = data.frame(dt1 = c("22DEC16", "12JUN16"), x = c(10,20))df = df %>% mutate(dt2 = as.Date(dt1, "%d%b%y"))>dfdt1 x dt21 22DEC16 10 2016-12-222 12JU
..
Mesos 上的独立 Spark 集群访问不同 Hadoop 集群中的 HDFS 数据
我们有一个包含 275 个节点的 Hadoop 集群(55Tb 总内存，12000 个 VCore).这个集群与几个项目共享，我们有一个 YARN 队列分配给我们，但资源有限. 为了增强性能，我们正在考虑为我们的项目构建一个单独的 Spark 集群(在同一网络中的 Mesos 上)并访问 Hadoop 集群上的 HDFS 数据. 如 Spark 文档中所述:https://spark.
..
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("分隔符", ",").optio
..
在数据框 spark 中查找属性组合的前 n 个元素
我有一个如下所示的数据框. Scala>ds.show+----+----------+----------+-----+|键|属性1|属性2|值|+----+----------+----------+-----+|mac1|A1|B1|10||mac2|A2|B1|10||mac3|A2|B1|10||mac1|A1|B2|10||mac1|A1|B2|10||mac3|A1|B1|10|
..
持久化比非持久化调用慢
我的设置是:Spark 2.1 在 160 GB、48 个 vcore 的 3 节点 YARN 集群上.动态分配开启.spark.executor.memory=6G, spark.executor.cores=6 首先，我正在阅读 hive 表:订单 (329MB) 和 lineitems (1.43GB) 和做左外连接.接下来，我根据加入的条件应用 7 种不同的过滤条件数据集(类似于 v
..
如何在 CASE 语句中使用数组类型列值
我有一个包含两列的数据框，listA 存储为 Seq[String] 和 valB 存储为 String>.我想创建第三列 valC，它将是 Int 类型，其值为 如果 valB 存在于 listA 中则为 1 否则为 0 我尝试执行以下操作: val dfWithAdditionalColumn = df.withColumn("valC", when($"listA".contain
..