python - Spark 属性错误 : Can't get attribute 'new_block' on

转载 作者：行者123 更新时间：2023-12-04 11:31:32 25 4

我在 AWS EMR 上使用 pyspark(4 r5.xlarge 作为 4 个 worker ，每个有一个执行程序和 4 个内核)，我得到了 AttributeError: Can't get attribute 'new_block' on <module 'pandas.core.internals.blocks' .以下是引发此错误的代码片段:

search =  SearchEngine(db_file_dir = "/tmp/db")
conn = sqlite3.connect("/tmp/db/simple_db.sqlite")
pdf_ = pd.read_sql_query('''select  zipcode, lat, lng, 
                        bounds_west, bounds_east, bounds_north, bounds_south from 
                        simple_zipcode''',conn)
brd_pdf = spark.sparkContext.broadcast(pdf_) 
conn.close()


@udf('string')
def get_zip_b(lat, lng):
    pdf = brd_pdf.value 
    out = pdf[(np.array(pdf["bounds_north"]) >= lat) & 
              (np.array(pdf["bounds_south"]) <= lat) & 
              (np.array(pdf['bounds_west']) <= lng) & 
              (np.array(pdf['bounds_east']) >= lng) ]
    if len(out):
        min_index = np.argmin( (np.array(out["lat"]) - lat)**2 + (np.array(out["lng"]) - lng)**2)
        zip_ = str(out["zipcode"].iloc[min_index])
    else:
        zip_ = 'bad'
    return zip_

df = df.withColumn('zipcode', get_zip_b(col("latitude"),col("longitude")))

下面是回溯，get_zip_b 中的第 102 行指的是 pdf = brd_pdf.value :

21/08/02 06:18:19 WARN TaskSetManager: Lost task 12.0 in stage 7.0 (TID 1814, ip-10-22-17-94.pclc0.merkle.local, executor 6): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/worker.py", line 605, in main
    process()
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/worker.py", line 597, in process
    serializer.dump_stream(out_iter, outfile)
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/serializers.py", line 223, in dump_stream
    self.serializer.dump_stream(self._batched(iterator), stream)
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/serializers.py", line 141, in dump_stream
    for obj in iterator:
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/serializers.py", line 212, in _batched
    for item in iterator:
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/worker.py", line 450, in mapper
    result = tuple(f(*[a[o] for o in arg_offsets]) for (arg_offsets, f) in udfs)
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/worker.py", line 450, in <genexpr>
    result = tuple(f(*[a[o] for o in arg_offsets]) for (arg_offsets, f) in udfs)
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/worker.py", line 90, in <lambda>
    return lambda *a: f(*a)
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/util.py", line 121, in wrapper
    return f(*args, **kwargs)
  File "/mnt/var/lib/hadoop/steps/s-1IBFS0SYWA19Z/Mobile_ID_process_center.py", line 102, in get_zip_b
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/broadcast.py", line 146, in value
    self._value = self.load_from_path(self._path)
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/broadcast.py", line 123, in load_from_path
    return self.load(f)
  File "/mnt/yarn/usercache/hadoop/appcache/application_1627867699893_0001/container_1627867699893_0001_01_000009/pyspark.zip/pyspark/broadcast.py", line 129, in load
    return pickle.load(file)
AttributeError: Can't get attribute 'new_block' on <module 'pandas.core.internals.blocks' from '/mnt/miniconda/lib/python3.9/site-packages/pandas/core/internals/blocks.py'>

一些观察和思考过程:
1、在网上查了一下，pyspark中的AttributeError好像是driver和worker的pandas版本不匹配导致的？
2，但是我在两个不同的数据集上运行了相同的代码，一个没有任何错误，而另一个没有，这看起来很奇怪和不确定，而且错误似乎不是由不匹配的 Pandas 版本引起的。否则，两个数据集都不会成功。
3，然后我再次在成功的数据集上运行相同的代码，但这次使用不同的 Spark 配置:将 spark.driver.memory 从 2048M 设置为 4192m，它抛出了 AttributeError。
4、总之，我认为AttributeError与驱动程序有关。但我无法从错误消息中判断它们之间的关系，以及如何修复它:AttributeError: Can't get attribute 'new_block' on

最佳答案

我在服务器中使用 pandas 1.3.2 而在我的客户端中使用 1.2 时遇到了同样的错误。
将 Pandas 降级到 1.2 解决了这个问题。

关于python - Spark 属性错误 : Can't get attribute 'new_block' on <module 'pandas.core.internals.blocks' ，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/68625748/

25 4 0

• c# - 无法从 Func 转换为 Func

  我对这个错误很困惑: Cannot implicitly convert type 'System.Func [c:\Program Files (x86)\Reference Assemblies\

• rust - 所有 T、&T、&mut T、[T]、&[T]、*mut T 的通用实现

  考虑这段代码: pub trait Hello { fn hello(&self); } impl Hello for Any { fn hello(&self) {

• c++ - "T t = {};"和 "T t{};"可以产生不同的结果吗？

  问题很简单。是否可以构造这样一个类型 T，对于它下面的两个变量声明会产生不同的结果？ T t1 = {}; T t2{}; 我已经研究 cppreference 和标准一个多小时了，我了解以下内容:

• java - Comparator中的compare(T,T)不能应用于(T,T)

  Intellij idea 给我这个错误:“Compare (T, T) in Comparator cannot be applied to (T, T)” 对于以下代码: public class

• php -\n\t\t\t\t\在php中的含义和用途是什么

  任何人都可以告诉我 : n\t\t\t\t\n\t\t\t 在以下来自和 dwr 服务的响应中的含义和用途是什么. \r\n\t\t\t \r\n\t\t\t

• c++ - "T a"、 "T a()"和 "T a=T()"之间有什么区别，其中 T 是一个类？

  让 T 成为一个 C++ 类。 下面三个指令在行为上有什么区别吗？ T a; T a(); T a = T(); T 为不带参数的构造函数提供了显式定义这一事实是否对问题有任何改变？ 后续问题:如果

• Rust中的智能指针:BoxRcArcCellRefCellWeak

  Rust中的智能指针是什么 智能指针（smart pointers）是一类数据结构，是拥有数据所有权和额外功能的指针。是指针的进一步发展 指针（pointer）是一个包含内存地

• c++ - 交替排序数组的最简单方法是什么(例如 :T, T,F,F,F,T,F,T,F,F,F 到 T,F,T,F,T,F,T,F,T,T ,T)?

  比如我有一个 vector vector > v={{true,1},{true,2},{false,3},{false,4},{false,5},{true,6},{false,7},{true,8

• Python 3 Pandas - 带有额外\n\t\t\t\t\t 的列

  我有一个来自 .xls 电子表格的数据框，我打印了 print(df.columns.values) 列，输出包含一个名为:Poll Responses\n\t\t\t\t\t。 我查看了 Excel

• java - 为什么Stream.reduce采用BinaryOperator 而不是BiFunction ？

  This question already has answers here: What are good reasons for choosing invariance in an API like

• rust - &T 和 T/&, ~T 和 T/~ 的区别

  指针类型作为类型前缀与在类型前加斜杠作为后缀有什么区别。斜线到底是什么意思？ 最佳答案 语法 T/~ 和 T/& 基本上已被弃用(我什至不确定编译器是否仍然接受它)。在向新向量方案过渡的初始阶段，[T

• c++ - 如何从 T、T& 或 T* 模板参数中获取 T*

  我正在尝试找到一种方法来获取模板参数的基类。 考虑以下类: template class Foo { public: Foo(){}; ~Foo(){};

• c++ - 当 T 声明为 T::T(T&) 时，std::is_trivially_constructible::value 的结果不正确

  这是一个让我感到困惑的小问题。我不知道如何描述它，所以只看下面的代码: struct B { B() {} B(B&) { std::cout ::value #include

• c++ - T::T(T&)有什么用？

  为什么有 T::T(T&) 而 T::T(const T&) 更适合 copy ？ (大概是用来实现move语义的？？？) 原始描述(被melpomene证明是错误的): 在C++11中，支持了一种新

• java - 类型参数 T 使用 Eclipse 在 T[] toArray(T[] a) 中隐藏类型 T

  在 Java 7 中使用 eclipse 4.2 并尝试实现 List 接口(interface)的以下方法时，我收到了警告。 public T[] toArray(T[] a) { ret

• foo[T](T,T) : T 的 Scala 类型扩展/推断

  假设有三个函数: def foo[T](a:T, b:T): T = a def test1 = foo(1, "2") def test2 = foo(List(), ListBuffer()) 虽

• generics - 通用类型统一 : multiple parameters (T, T) 与多个参数列表 (T)(T)？

  我对柯里化(Currying)和非柯里化(Currying)泛型函数之间类型检查的差异有点困惑: scala> def x[T](a: T, b: T) = (a == b) x: [T](a: T,

• c++ - 转发引用 : returning T when given T&& and T& when given T&

  考虑一个类A，我如何编写一个具有与相同行为的模板 A& pretty(A& x) { /* make x pretty */ return x; } A pretty(A&& x) {

• java - 为什么在 T 是类型参数而 t 是变量的情况下不允许使用 "t instanceof T"？

  Eclipse 表示由于泛型类型橡皮擦，类型参数不允许使用 instanceof 操作。 我同意在运行时不会保留任何类型信息。但是请考虑以下类的通用声明: class SomeClass{ T

• c++ - T t {x} 和 T t = { x } 对于整数或枚举类型的区别？

  在 C++14 中: 对于任何整数或枚举类型 T 以及对于任何表达式 expr: 有没有区别: struct S { T t { expr }; }; 和 struct S { T t = { exp