个人中心
文章发布
退出
作者热门文章
- android - 多次调用 OnPrimaryClipChangedListener
- android - 无法更新 RecyclerView 中的 TextView 字段
- android.database.CursorIndexOutOfBoundsException : Index 0 requested, 光标大小为 0
- android - 使用 AppCompat 时，我们是否需要明确指定其 UI 组件(Spinner、EditText)颜色
25
4
目前，在处理复杂的嵌套结构时，Spark 结构和 Dask 之间的集成似乎比较麻烦。尽管 parquet 加载是一项大型持续工作的一部分(fastparquet、pyarrow)，但专门转储具有由 Dask 读取的嵌套结构的 Spark Dataframe 似乎还不是很可靠；
所以我的后续问题 - 假设我可以接受在 Spark 中进行一些转换，并将 DataFrame 转换为包含自定义类对象的 RDD；有没有一种方法可以可靠地转储具有自定义类对象的 Spark RDD 的数据并在 Dask 集合中读取它？显然，您可以将 rdd 收集到 python 列表中，对其进行 pickle，然后将其作为普通数据结构读取，但这消除了加载大于内存数据集的机会。 dask 可以使用 Spark pickle 之类的东西来加载分布式 pickle 吗？
最佳答案
我通过执行以下操作解决了这个问题
拥有一个包含自定义对象列表作为行值的 Spark RDD，我创建了一个版本的 rdd，其中使用 cPickle.dumps 将对象序列化为字符串。然后将此 RDD 转换为带有字符串列的简单 DF 并将其写入 parquet。 Dask能够读取结构简单的parquet文件。然后使用cPickle.loads反序列化以获得原始对象
关于python - 有没有什么好方法将Spark RDD的内容读取到Dask结构中，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/53169690/
25
4
0
我是 Pyspark 新手，我使用的是 Spark 2.0.2。 我有一个名为 Test_RDD 的 RDD，其结构如下: U-Key || V1 || V2 || V3 || ----
我正在寻找一种方法将一个 RDD 拆分为两个或多个 RDD，并将获得的结果保存为两个单独的 RDD。例如: rdd_test = sc.parallelize(range(50), 1) 我的代码:
我有一个结构如下的RDD: ((user_id,item_id,rating)) 让我们将此 RDD 称为训练 然后还有另一个具有相同结构的rdd: ((user_id,item_id,rating)
已经有人问过类似的问题。最相似的是这个: Spark: How to split an RDD[T]` into Seq[RDD[T]] and preserve the ordering 但是，我不
我正在使用 spark 来处理数据。但是我不知道如何将新数据保存到Hive 我从 Hive 加载 rdd，然后运行 map 函数来清理数据。 result = myRdd.map(lambda x
我有一个名为 index 的 rdd:RDD[(String, String)]，我想用 index 来处理我的文件。 这是代码: val get = file.map({x => val tmp
我有两个 RDD: **rdd1** id1 val1 id2 val2 **rdd2** id1 v1 id2 v2 id1 v3 id8 v7 id1 v4 id3 v5 id6 v6 我想过滤
我有一个 RDD，需要从另一个 RDD 访问数据。但是，我总是收到任务不可序列化错误。我已经扩展了 Serialized 类，但它没有起作用。代码是: val oldError = rddOfRati
我有一个 RDD，需要从另一个 RDD 访问数据。但是，我总是收到任务不可序列化错误。我已经扩展了 Serialized 类，但它没有起作用。代码是: val oldError = rddOfRati
我有一个 RDD 对: (105,918) (105,757) (502,516) (105,137) (516,816) (350,502) 我想将它分成两个 RDD，这样第一个只有具有非重复值的对
我正在尝试使用 spark 执行 K 最近邻搜索。 我有一个 RDD[Seq[Double]] 并且我打算返回一个 RDD[(Seq[Double],Seq[Seq[Double]])] 带有实际行和
我是Spark和Scala语言的新手，并且希望将所有RDD合并到一个List中，如下所示(List to RDD): val data = for (item {
我找不到只参与 rdd 的方法. take看起来很有希望，但它返回 list而不是 rdd .我当然可以将其转换为 rdd ，但这似乎既浪费又丑陋。 my_rdd = sc.textFile("my
我正在寻找一种将 RDD 拆分为两个或更多 RDD 的方法。我见过的最接近的是 Scala Spark: Split collection into several RDD?这仍然是一个单一的 RDD
我有一个RDD[String]，wordRDD。我还有一个从字符串/单词创建 RDD[String] 的函数。我想为 wordRDD 中的每个字符串创建一个新的 RDD。以下是我的尝试: 1) 失败，
我刚刚开始使用 Spark 和 Scala 我有一个包含多个文件的目录我使用 成功加载它们 sc.wholeTextFiles(directory) 现在我想升一级。我实际上有一个目录，其中包含包含文
我想从另一个 RDD 中减去一个 RDD。我查看了文档，发现 subtract可以这样做。实际上，当我测试时 subtract , 最终的 RDD 保持不变，值不会被删除! 有没有其他功能可以做到这一
我在 HDFS 中有如下三个文件中的数据 EmployeeManagers.txt (EmpID,ManagerID) 1,5 2,4 3,4 4,6 5,6 EmployeeNames.txt (E
我正在开发一个应用程序，我需要对 RDD 中具有相同键的每对行执行计算，这是 RDD 结构: List>> dat2 = new ArrayList<>(); dat2.add(new Tuple2>
我在 spark 集群中有两个文件，foo.csv 和 bar.csv，它们都有 4 列和完全相同的字段:时间、用户、url、类别。 我想通过 bar.csv 的某些列过滤掉 foo.csv。最后，我
我是一名优秀的程序员,十分优秀！