function `aggregate` NotImplementedError

[hongzhouye]

When I tried to run the following simple script by python xxx.py for sanity check

from pyspark import SparkContext
from drudge import PartHoleDrudge

ctx = SparkContext()
dr = PartHoleDrudge(ctx)

which is taken from the conf_ph.py from the tutorial. I got error msg:

Traceback (most recent call last):
  File "test.py", line 5, in <module>
    dr = PartHoleDrudge(ctx)
  File "/Users/hzye/local/opt/miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/fock.py", line 816, in __init__
    **kwargs)
  File "/Users/hzye/local/opt/miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/fock.py", line 750, in __init__
    orig_ham = (one_body_ham + two_body_ham).reset_dumms()
  File "/Users/hzye/local/opt/miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py", line 417, in reset_dumms
    free_vars = self.free_vars
  File "/Users/hzye/local/opt/miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py", line 211, in free_vars
    self._free_vars = self._get_free_vars(self._terms)
  File "/Users/hzye/local/opt/miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py", line 224, in _get_free_vars
    ).aggregate(set(), _union, _union)
  File "/Users/hzye/local/opt/miniconda/envs/frank/lib/python3.6/site-packages/dummy_spark/rdd.py", line 304, in aggregate
    raise NotImplementedError
NotImplementedError

I checked the source code rdd.py from dummy_spark, and it seems that the function aggregate is indeed not implemented therein but being called by drudge.py. Does anyone know how this could be resolved?

Thanks in advance!

[chenpeizhi]

To use dummy_spark, you should do

from dummy_spark import SparkContext

instead of

from pyspark import SparkContext

If you import from pyspark, the pyspark package should have been installed in your Python environment.

[hongzhouye]

@chenpeizhi sorry that was a typo. I got the error message when using

from dummy_spark import SparkContext

[chenpeizhi]

@hongzhouye I see. Are you using the original DummyRDD package or the fork by Jinmo?

[hongzhouye]

@chenpeizhi I used the one from PyPI. After switching to Jinmo's fork, no error was seen anymore. That also makes the ccd.drs script run successfully from this tutorial. So this seems to be resolved. Thanks!

However, when I tried to run the "advanced ccd" example from the same tutorial, i got segfault for the optimize step, i.e.,

# When the gristmill package is also installed, the evaluation of the working
# equations can also be optimized with it.
eval_seq = optimize(
    [e, r2], substs={no: 1000, nv: 10000}
)

I had gristmill installed on my laptop. But when i run the tests in gristmill, they all failed except for conftest.py (which is the trivial one). All the error messages seem to have the following

../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:203: in is_scalar
    self._terms.map(lambda x: x.is_scalar).collect()
../../miniconda/envs/frank/lib/python3.6/site-packages/pyspark-2.4.5-py3.6.egg/pyspark/rdd.py:816: in collect
    sock_info = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())

which says the drudge.py script calls the rdd.py from pystark. So I checked drudge.py and found this line of importing

from pyspark import RDD, SparkContext

Now comes my question: is this import meant to import from the original pystark package or some forked version from the authors of drudge?

Thanks a lot!

[gauravharsha]

There is only one version of pyspark package that drudge uses, and it is the same as the one that is available on PyPI. On the other hand, the dummy_spark from PyPI is not complete (for instance, as you have discovered above, collect is a pyspark function, but a serialized implementation is missing from the official dummy_spark release). So, you need to use the fork from Jinmo (as @chenpeizhi explained above).

The import statements in drudge.py are merely to define the datatypes of various objects defined therein. The SparkContext instance that you provide as an argument while initializing the drudge (say the PartHoleDrudge) would eventually define whether it uses a pyspark (parallel) or a dummy_spark (serialized) version.

Could you share where exactly in the tests do the above errors arise. That would certainly help identify the source; the tests and the "advanced ccd" tutorial run fine on my machine.

[hongzhouye]

@gauravharsha

There is only one version of pyspark package that drudge uses, and it is the same as the one that is available on PyPI. On the other hand, the dummy_spark from PyPI is not complete (for instance, as you have discovered above, collect is a pyspark function, but a serialized implementation is missing from the official dummy_spark release). So, you need to use the fork from Jinmo (as @chenpeizhi explained above).

Yep, switching to Jinmo's dummy_spark indeed resolves the original issue! Thanks!

Could you share where exactly in the tests do the above errors arise. That would certainly help identify the source; the tests and the "advanced ccd" tutorial run fine on my machine.

The error I was referring to was seen when running tests in gristmill/tests. The reason I traced to gristmill was that the "advanced ccd" script runs fine until the optimize step, which is from gristmill based on my understanding. As I said, all tests in gristmill/tests failed with similar error messages (except for the trivial test conftest.py). To be specific, I included here the full error message for the test function test_matrix_chain from gristmill/tests/opt_matrix_test.py:

________________________________________ test_matrix_chain _________________________________________

three_ranges = <drudge.drudge.Drudge object at 0xa17199cf8>

    def test_matrix_chain(three_ranges):
        """Test a basic matrix chain multiplication problem.
    
        Here a very simple matrix chain multiplication problem with three
        matrices are used to test the factorization facilities.  In this simple
        test, we will have three matrices :math:`x`, :math:`y`, and :math:`z`,
        which are of shapes :math:`m\\times n`, :math:`n \\times l`, and :math:`l
        \\times m` respectively. In the factorization, we are going to set
        :math:`n = 2 m` and :math:`l = 3 m`.
    
        """
    
        dr = three_ranges
        p = dr.names
        m, n, l = p.m, p.n, p.l
    
        # The indexed bases.
        x = IndexedBase('x', shape=(m, n))
        y = IndexedBase('y', shape=(n, l))
        z = IndexedBase('z', shape=(l, m))
    
        target_base = IndexedBase('t')
        target = dr.define_einst(
            target_base[p.a, p.b],
>           x[p.a, p.i] * y[p.i, p.p] * z[p.p, p.b]
        )

opt_matrix_test.py:81: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:2785: in define_einst
    return TensorDef(base, exts, tensor)
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:1761: in __init__
    is_scalar = self.is_scalar
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:203: in is_scalar
    self._terms.map(lambda x: x.is_scalar).collect()
../../miniconda/envs/frank/lib/python3.6/site-packages/pyspark-2.4.5-py3.6.egg/pyspark/rdd.py:816: in collect
    sock_info = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
../../miniconda/envs/frank/lib/python3.6/site-packages/py4j-0.10.7-py3.6.egg/py4j/java_gateway.py:1257: in __call__
    answer, self.gateway_client, self.target_id, self.name)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

answer = 'xro36', gateway_client = <py4j.java_gateway.GatewayClient object at 0xa17232ba8>
target_id = 'z:org.apache.spark.api.python.PythonRDD', name = 'collectAndServe'

    def get_return_value(answer, gateway_client, target_id=None, name=None):
        """Converts an answer received from the Java gateway into a Python object.
    
        For example, string representation of integers are converted to Python
        integer, string representation of objects are converted to JavaObject
        instances, etc.
    
        :param answer: the string returned by the Java gateway
        :param gateway_client: the gateway client used to communicate with the Java
            Gateway. Only necessary if the answer is a reference (e.g., object,
            list, map)
        :param target_id: the name of the object from which the answer comes from
            (e.g., *object1* in `object1.hello()`). Optional.
        :param name: the name of the member from which the answer comes from
            (e.g., *hello* in `object1.hello()`). Optional.
        """
        if is_error(answer)[0]:
            if len(answer) > 1:
                type = answer[1]
                value = OUTPUT_CONVERTER[type](answer[2:], gateway_client)
                if answer[1] == REFERENCE_TYPE:
                    raise Py4JJavaError(
                        "An error occurred while calling {0}{1}{2}.\n".
>                       format(target_id, ".", name), value)
E                   py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
E                   : java.lang.IllegalArgumentException: Unsupported class file major version 56
E                   	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:166)
E                   	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:148)
E                   	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:136)
E                   	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:237)
E                   	at org.apache.spark.util.ClosureCleaner$.getClassReader(ClosureCleaner.scala:49)
E                   	at org.apache.spark.util.FieldAccessFinder$$anon$3$$anonfun$visitMethodInsn$2.apply(ClosureCleaner.scala:517)
E                   	at org.apache.spark.util.FieldAccessFinder$$anon$3$$anonfun$visitMethodInsn$2.apply(ClosureCleaner.scala:500)
E                   	at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
E                   	at scala.collection.mutable.HashMap$$anon$1$$anonfun$foreach$2.apply(HashMap.scala:134)
E                   	at scala.collection.mutable.HashMap$$anon$1$$anonfun$foreach$2.apply(HashMap.scala:134)
E                   	at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:236)
E                   	at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
E                   	at scala.collection.mutable.HashMap$$anon$1.foreach(HashMap.scala:134)
E                   	at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
E                   	at org.apache.spark.util.FieldAccessFinder$$anon$3.visitMethodInsn(ClosureCleaner.scala:500)
E                   	at org.apache.xbean.asm6.ClassReader.readCode(ClassReader.java:2175)
E                   	at org.apache.xbean.asm6.ClassReader.readMethod(ClassReader.java:1238)
E                   	at org.apache.xbean.asm6.ClassReader.accept(ClassReader.java:631)
E                   	at org.apache.xbean.asm6.ClassReader.accept(ClassReader.java:355)
E                   	at org.apache.spark.util.ClosureCleaner$$anonfun$org$apache$spark$util$ClosureCleaner$$clean$14.apply(ClosureCleaner.scala:307)
E                   	at org.apache.spark.util.ClosureCleaner$$anonfun$org$apache$spark$util$ClosureCleaner$$clean$14.apply(ClosureCleaner.scala:306)
E                   	at scala.collection.immutable.List.foreach(List.scala:392)
E                   	at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:306)
E                   	at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:162)
E                   	at org.apache.spark.SparkContext.clean(SparkContext.scala:2326)
E                   	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2100)
E                   	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2126)
E                   	at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:990)
E                   	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
E                   	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
E                   	at org.apache.spark.rdd.RDD.withScope(RDD.scala:385)
E                   	at org.apache.spark.rdd.RDD.collect(RDD.scala:989)
E                   	at org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:166)
E                   	at org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)
E                   	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
E                   	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
E                   	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
E                   	at java.base/java.lang.reflect.Method.invoke(Method.java:567)
E                   	at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
E                   	at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
E                   	at py4j.Gateway.invoke(Gateway.java:282)
E                   	at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
E                   	at py4j.commands.CallCommand.execute(CallCommand.java:79)
E                   	at py4j.GatewayConnection.run(GatewayConnection.java:238)
E                   	at java.base/java.lang.Thread.run(Thread.java:835)

../../miniconda/envs/frank/lib/python3.6/site-packages/py4j-0.10.7-py3.6.egg/py4j/protocol.py:328: Py4JJavaError
-------------------------------------- Captured stderr setup ---------------------------------------
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/Users/hzye/local/opt/miniconda/envs/frank/lib/python3.6/site-packages/pyspark-2.4.5-py3.6.egg/pyspark/jars/spark-unsafe_2.11-2.4.5.jar) to method java.nio.Bits.unaligned()
WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
20/05/07 09:10:24 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).

Note that I saw similar error messages when running some tests in drudge/tests, too. For example, the error message below came from running the test test_bogoliubov_has_hamiltonian from drudge/tests/bogoliubov_test.py:

________________________ ERROR at setup of test_bogoliubov_has_hamiltonian _________________________

spark_ctx = <SparkContext master=local[2] appName=drudge-unittest>

    @pytest.fixture(scope='module')
    def bogoliubov(spark_ctx):
        """Initialize the environment for a free algebra."""
    
>       dr = BogoliubovDrudge(spark_ctx)

bogoliubov_test.py:16: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/fock.py:1130: in __init__
    one_body=one_body, two_body=two_body, dbbar=dbbar, **kwargs
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/fock.py:750: in __init__
    orig_ham = (one_body_ham + two_body_ham).reset_dumms()
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:417: in reset_dumms
    free_vars = self.free_vars
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:211: in free_vars
    self._free_vars = self._get_free_vars(self._terms)
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:224: in _get_free_vars
    ).aggregate(set(), _union, _union)
../../miniconda/envs/frank/lib/python3.6/site-packages/pyspark-2.4.5-py3.6.egg/pyspark/rdd.py:952: in aggregate
    vals = self.mapPartitions(func).collect()
../../miniconda/envs/frank/lib/python3.6/site-packages/pyspark-2.4.5-py3.6.egg/pyspark/rdd.py:816: in collect
    sock_info = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
../../miniconda/envs/frank/lib/python3.6/site-packages/py4j-0.10.7-py3.6.egg/py4j/java_gateway.py:1257: in __call__
    answer, self.gateway_client, self.target_id, self.name)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

answer = 'xro39', gateway_client = <py4j.java_gateway.GatewayClient object at 0x119daf208>
target_id = 'z:org.apache.spark.api.python.PythonRDD', name = 'collectAndServe'

    def get_return_value(answer, gateway_client, target_id=None, name=None):
        """Converts an answer received from the Java gateway into a Python object.
    
        For example, string representation of integers are converted to Python
        integer, string representation of objects are converted to JavaObject
        instances, etc.
    
        :param answer: the string returned by the Java gateway
        :param gateway_client: the gateway client used to communicate with the Java
            Gateway. Only necessary if the answer is a reference (e.g., object,
            list, map)
        :param target_id: the name of the object from which the answer comes from
            (e.g., *object1* in `object1.hello()`). Optional.
        :param name: the name of the member from which the answer comes from
            (e.g., *hello* in `object1.hello()`). Optional.
        """
        if is_error(answer)[0]:
            if len(answer) > 1:
                type = answer[1]
                value = OUTPUT_CONVERTER[type](answer[2:], gateway_client)
                if answer[1] == REFERENCE_TYPE:
                    raise Py4JJavaError(
                        "An error occurred while calling {0}{1}{2}.\n".
>                       format(target_id, ".", name), value)
E                   py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
E                   : java.lang.IllegalArgumentException: Unsupported class file major version 56
E                   	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:166)
E                   	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:148)
E                   	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:136)
E                   	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:237)
E                   	at org.apache.spark.util.ClosureCleaner$.getClassReader(ClosureCleaner.scala:49)
E                   	at org.apache.spark.util.FieldAccessFinder$$anon$3$$anonfun$visitMethodInsn$2.apply(ClosureCleaner.scala:517)
E                   	at org.apache.spark.util.FieldAccessFinder$$anon$3$$anonfun$visitMethodInsn$2.apply(ClosureCleaner.scala:500)
E                   	at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
E                   	at scala.collection.mutable.HashMap$$anon$1$$anonfun$foreach$2.apply(HashMap.scala:134)
E                   	at scala.collection.mutable.HashMap$$anon$1$$anonfun$foreach$2.apply(HashMap.scala:134)
E                   	at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:236)
E                   	at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
E                   	at scala.collection.mutable.HashMap$$anon$1.foreach(HashMap.scala:134)
E                   	at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
E                   	at org.apache.spark.util.FieldAccessFinder$$anon$3.visitMethodInsn(ClosureCleaner.scala:500)
E                   	at org.apache.xbean.asm6.ClassReader.readCode(ClassReader.java:2175)
E                   	at org.apache.xbean.asm6.ClassReader.readMethod(ClassReader.java:1238)
E                   	at org.apache.xbean.asm6.ClassReader.accept(ClassReader.java:631)
E                   	at org.apache.xbean.asm6.ClassReader.accept(ClassReader.java:355)
E                   	at org.apache.spark.util.ClosureCleaner$$anonfun$org$apache$spark$util$ClosureCleaner$$clean$14.apply(ClosureCleaner.scala:307)
E                   	at org.apache.spark.util.ClosureCleaner$$anonfun$org$apache$spark$util$ClosureCleaner$$clean$14.apply(ClosureCleaner.scala:306)
E                   	at scala.collection.immutable.List.foreach(List.scala:392)
E                   	at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:306)
E                   	at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:162)
E                   	at org.apache.spark.SparkContext.clean(SparkContext.scala:2326)
E                   	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2100)
E                   	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2126)
E                   	at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:990)
E                   	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
E                   	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
E                   	at org.apache.spark.rdd.RDD.withScope(RDD.scala:385)
E                   	at org.apache.spark.rdd.RDD.collect(RDD.scala:989)
E                   	at org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:166)
E                   	at org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)
E                   	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
E                   	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
E                   	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
E                   	at java.base/java.lang.reflect.Method.invoke(Method.java:567)
E                   	at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
E                   	at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
E                   	at py4j.Gateway.invoke(Gateway.java:282)
E                   	at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
E                   	at py4j.commands.CallCommand.execute(CallCommand.java:79)
E                   	at py4j.GatewayConnection.run(GatewayConnection.java:238)
E                   	at java.base/java.lang.Thread.run(Thread.java:835)

../../miniconda/envs/frank/lib/python3.6/site-packages/py4j-0.10.7-py3.6.egg/py4j/protocol.py:328: Py4JJavaError
-------------------------------------- Captured stderr setup ---------------------------------------
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/Users/hzye/local/opt/miniconda/envs/frank/lib/python3.6/site-packages/pyspark-2.4.5-py3.6.egg/pyspark/jars/spark-unsafe_2.11-2.4.5.jar) to method java.nio.Bits.unaligned()
WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
20/05/07 09:07:05 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).

The error seems to occur at the stage of initializing the environment, i.e.

dr = BogoliubovDrudge(spark_ctx)

I suspect the problem is with my pyspark(which I installed via pip). Because when I ran python xxx.py with the following script

from dummy_spark import SparkContext               # Jinmo's dummy_spark
from drudge import ReducedBCSDrudge

ctx = SparkContext()
dr = ReducedBCSDrudge(ctx)

It gives no error, which seems to suggest at least those different Drudge's work well with Jinmo's dummy_spark...

[chenpeizhi]

@hongzhouye

Now comes my question: is this import meant to import from the original pystark package or some forked version from the authors of drudge?

We use the origional pyspark. On my computer, I use Python 3.7.4 and pyspark 2.4.3. We haven't tested pyspark 2.4.5 yet. It might be worth testing the older pyspark 2.4.3 on your computer to see whether anything changes.

The reason I traced to gristmill was that the "advanced ccd" script runs fine until the optimize step, which is from gristmill based on my understanding.

optimize is indeed a gristmill function, but it takes as input a list of TensorDef instances, which are drudge objects. By default, TensorDef.simplify is called, which will invoke pyspark or dummy_spark.

[hongzhouye]

@chenpeizhi Thanks for your response!

We use the origional pyspark. On my computer, I use Python 3.7.4 and pyspark 2.4.3. We haven't tested pyspark 2.4.5 yet. It might be worth testing the older pyspark 2.4.3 on your computer to see whether anything changes.

I tried uninstall 2.4.5 and reinstall pyspark 2.4.3 and the problem is still there. Now i am more convinced that the problem is with the pyspark. For example, in command line by pyspark i can successfully log in and see the following prompt

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.4.3
      /_/

Using Python version 3.6.9 (default, Jul 30 2019 13:42:17)
SparkSession available as 'spark'.
>>>

But when I tried to run some simple examples like estimating pi in the pyspark prompt

import random
NUM_SAMPLES = 100000000
def inside(p):
    x, y = random.random(), random.random()
    return x*x + y*y < 1
count = sc.parallelize(range(0, NUM_SAMPLES)).filter(inside).count()
pi = 4 * count / NUM_SAMPLES
print(“Pi is roughly”, pi)

it failed at the count = sc.parallelize(...) line with an error message that is very similar to what I saw above:

Traceback (most recent call last):
  File "/Users/hzye/local/opt/miniconda/envs/frank/lib/python3.6/site-packages/pyspark/sql/utils.py", line 63, in deco
    return f(*a, **kw)
  File "/Users/hzye/local/opt/miniconda/envs/frank/lib/python3.6/site-packages/pyspark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py", line 328, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: java.lang.IllegalArgumentException: Unsupported class file major version 56
	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:166)
	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:148)
	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:136)
	at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:237)
	at org.apache.spark.util.ClosureCleaner$.getClassReader(ClosureCleaner.scala:49)
...

Since you mentioned that

optimize is indeed a gristmill function, but it takes as input a list of TensorDef instances, which are drudge objects. By default, TensorDef.simplify is called, which will invoke pyspark or dummy_spark.

is it possible to specifically ask the optimize function to use Jinmo's dummy_spark as the backend (as that seems to work according to my successful run of the naïve ccd.drs script). I don't mind if it runs in serial.

[chenpeizhi]

is it possible to specifically ask the optimize function to use Jinmo's dummy_spark as the backend (as that seems to work according to my successful run of the naïve ccd.drs script). I don't mind if it runs in serial.

You may do

export DUMMY_SPARK=1

before running the gristmill tests.

[gauravharsha]

There appears to be some incompatibility in Java/Spark. A simple way to avoid them would be to either use dummy_spark or alternatively use Docker images.

[hongzhouye]

@chenpeizhi

I tried this and rerun the "advanced ccd" tutorial script, it still failed with a segfault. I then tried rerunning the tests from gristmill/tests. I did not get the pyspark error messages as above. Instead I got a segfault which is likely to be the same as the one caused by the optimize function call in the "advanced ccd" tutorial.

Good news is that when I reran the tests in drudge/tests, most of them now pass (previously most of them failed). The three that failed are

nuclear_test.py::test_varsh_872_5
nuclear_test.py::test_wigner3j_sum_to_wigner6j
spin_one_half_test.py::test_restricted_parthole_drudge_simplification

with the following failure messages

• For test_varsh_872_5

nuclear = <drudge.nuclear.NuclearBogoliubovDrudge object at 0xa0ed48cf8>

    @skip_in_spark(reason="Maybe assumptions are not serialized with symbols")
    def test_varsh_872_5(nuclear: NuclearBogoliubovDrudge):
        """Test simplification based on the rule in Varshalovich 8.7.2 Eq (5).
    
        TODO: Investigate its failure in Apache Spark environment.
        """
        dr = nuclear
        a, alpha, b, beta, b_prm, beta_prm = symbols(
            'a alpha b beta bprm betaprm', integer=True
        )
        c, gamma = symbols('c gamma', integer=True)
        sums = [
            (alpha, Range('m', -a, a + 1)),
            (gamma, Range('M', -c, c + 1))
        ]
        amp = CG(a, alpha, b, beta, c, gamma) * CG(
            a, alpha, b_prm, beta_prm, c, gamma
        )
    
        expected = (
                KroneckerDelta(b, b_prm) * KroneckerDelta(beta, beta_prm)
                * (2 * c + 1) / (2 * b + 1)
        )
        for sums_i in [sums, reversed(sums)]:
            tensor = dr.sum(*sums_i, amp)
>           res = tensor.deep_simplify().merge()

nuclear_test.py:135: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/nuclear.py:437: in deep_simplify
    res = tensor.simplify()
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:708: in simplify
    self._drudge, self._simplify(self._terms), expanded=True
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:736: in _simplify
    terms = self._simplify_amps(terms)
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:456: in _simplify_amps
    lambda term: term.map(lambda x: x.simplify(), skip_vecs=True)
../../miniconda/envs/frank/lib/python3.6/site-packages/dummyrdd-0.0.7-py3.6.egg/dummy_spark/rdd.py:151: in map
    data = list(map(f, self._jrdd))
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:456: in <lambda>
    lambda term: term.map(lambda x: x.simplify(), skip_vecs=True)
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/term.py:801: in map
    res_amp = func(self._amp if amp is None else amp)
../../miniconda/envs/frank/lib/python3.6/site-packages/drudge-0.10.0.dev0-py3.6-macosx-10.7-x86_64.egg/drudge/drudge.py:456: in <lambda>
    lambda term: term.map(lambda x: x.simplify(), skip_vecs=True)
../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/core/expr.py:3497: in simplify
    return simplify(self, **kwargs)
../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/simplify/simplify.py:717: in simplify
    return done(expr)
../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/simplify/simplify.py:541: in done
    rv = e.doit() if doit else e
../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/core/basic.py:1691: in doit
    for term in self.args]
../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/core/basic.py:1691: in <listcomp>
    for term in self.args]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = CG(a, alpha, b, beta, c, gamma), hints = {}

    def doit(self, **hints):
        if self.is_symbolic:
>           raise ValueError("Coefficients must be numerical")
E           ValueError: Coefficients must be numerical

../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/physics/quantum/cg.py:194: ValueError

• For test_wigner3j_sum_to_wigner6j:

_________________________________ test_wigner3j_sum_to_wigner6j __________________________________

nuclear = <drudge.nuclear.NuclearBogoliubovDrudge object at 0xa0ed48cf8>

    def test_wigner3j_sum_to_wigner6j(nuclear: NuclearBogoliubovDrudge):
        """Test simplification of sum of product of four 3j's to a 6j.
    
        This test tries to simplify the original LHS of the equation from the
        Wolfram website.
        """
    
        dr = nuclear
        j1, j2, j3, jprm3, j4, j5, j6 = symbols(
            'j1 j2 j3 jprm3 j4 j5 j6', integer=True
        )
        m1, m2, m3, mprm3, m4, m5, m6 = symbols(
            'm1 m2 m3 mprm3 m4 m5 m6', integer=True
        )
    
        m_range = Range('m')
        sums = [(m_i, m_range[-j_i, j_i + 1]) for m_i, j_i in [
            (m1, j1), (m2, j2), (m4, j4), (m5, j5), (m6, j6)
        ]]
    
        phase = (-1) ** (
                j1 + j2 + j4 + j5 + j6 - m1 - m2 - m4 - m5 - m6
        )
        amp = (
                Wigner3j(j2, m2, j3, -m3, j1, m1)
                * Wigner3j(j1, -m1, j5, m5, j6, m6)
                * Wigner3j(j5, -m5, jprm3, mprm3, j4, m4)
                * Wigner3j(j4, -m4, j2, -m2, j6, -m6)
        )
    
        expected = (
                ((-1) ** (j3 - m3) / (2 * j3 + 1))
                * KroneckerDelta(j3, jprm3) * KroneckerDelta(m3, mprm3)
>               * Wigner6j(j1, j2, j3, j4, j5, j6)
        ).expand().simplify()

nuclear_test.py:210: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/core/expr.py:3497: in simplify
    return simplify(self, **kwargs)
../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/simplify/simplify.py:717: in simplify
    return done(expr)
../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/simplify/simplify.py:541: in done
    rv = e.doit() if doit else e
../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/core/basic.py:1691: in doit
    for term in self.args]
../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/core/basic.py:1691: in <listcomp>
    for term in self.args]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Wigner6j(j1, j2, j3, j4, j5, j6), hints = {}

    def doit(self, **hints):
        if self.is_symbolic:
>           raise ValueError("Coefficients must be numerical")
E           ValueError: Coefficients must be numerical

../../miniconda/envs/frank/lib/python3.6/site-packages/sympy/physics/quantum/cg.py:306: ValueError

• For test_restricted_parthole_drudge_simplification

_________________________ test_restricted_parthole_drudge_simplification _________________________

restricted_parthole = <drudge.fock.RestrictedPartHoleDrudge object at 0xa0ef62780>

    def test_restricted_parthole_drudge_simplification(restricted_parthole):
        """Test simplification in restricted particle-hole drudge.
    
        The purpose of this test is mostly on testing the correct resolution of
        ranges of constants up and down.
        """
    
        dr = restricted_parthole
        p = dr.names
        a = p.a
        sigma = dr.spin_dumms[0]
    
        op1 = dr.sum(p.c_[a, UP])
        op2 = dr.sum(p.c_dag[a, UP])
        res_concr = (op1 * op2 + op2 * op1).simplify()
    
        op1 = dr.sum((sigma, dr.spin_range), p.c_[a, sigma])
        res_abstr = (op1 * op2 + op2 * op1).simplify()
    
        for i in [res_concr, res_abstr]:
>           assert (i - Integer(1)).simplify() == 0
E           assert <drudge.drudg...t 0x10cc28ba0> == 0
E             -<drudge.drudge.Tensor object at 0x10cc28ba0>
E             +0

spin_one_half_test.py:161: AssertionError

You can see that they are all related to a call of the simplify function. However, what's weird is that in other tests there are also calls of the simplify function. Those seem fine though... Any ideas what is special about the three calls of simplify above?

[hongzhouye]

@gauravharsha

There appears to be some incompatibility in Java/Spark. A simple way to avoid them would be to either use dummy_spark or alternatively use Docker images.

I also saw discussions about the conflict between Java and Spark. Some suggest to downgrade java to java8 (I am using java12 now), which I unfortunately cannot do since I have other programs depending on java. BTW what is the version of java on your machine? unlikely that the error is due to this but just want to make sure :D

[chenpeizhi]

The DUMMY_SPARK environment variable only affects the tests (check conftest.py). To use dummy_spark for a python script, just import dummy_spark instead of pyspark. I was under the impression that you didn't get the segfault if dummy_spark was used for the tutorial. If dummy_spark also lead to a segfault, that's a very severe problem that we should look into.

The nuclear_test was causing problems. If I remember correctly, it has to do with the Wigner 3j and 6j symbol implementations in sympy; I think it got modified in the last few sympy releases. For the other failed cases related to simplify, Drudge managed to simplify the tensor to some simple terms, but failed to recognize that they are indeed 0. The Scuseria group is keeping track of these failed cases. We'll fix them in the next release.

[gauravharsha]

Second @chenpeizhi's comments.

I am using openjdk 1.8.0_252, which is Java 8. Also, as pointed out above, if you want to use dummy_spark all the way, all you need to do is replace pyspark with dummy_spark in the drudge configuration file.

As for

You can see that they are all related to a call of the simplify function. However, what's weird is that in other tests there are also calls of the simplify function...

each submodule in Drudge has its own rules for simplify, so you would expect them to work differently. Besides, if there is a problem with pyspark, you may want to run the tests for drudge using dummy_spark, which I think is true in your case.

[hongzhouye]

@chenpeizhi

The DUMMY_SPARK environment variable only affects the tests (check conftest.py). To use dummy_spark for a python script, just import dummy_spark instead of pyspark. I was under the impression that you didn't get the segfault if dummy_spark was used for the tutorial. If dummy_spark also lead to a segfault, that's a very severe problem that we should look into.

Sorry I did not make this clear. After using dummy_spark, the naïve ccd.drs tutorial runs fine, but the advanced on ccd_adv.drs failed with a segfault. The full output is

# in command line : python -m drudge conf_ph.py ccd_adv.drs 
Memoize: read data from h_bar.pickle
H-bar has 76 terms
Working equation derived!
zsh: segmentation fault  python -m drudge conf_ph.py rccd_advanced.py

You can see that it successfully derived the working equations but failed to optimize them.

Just to be super clear, the output above was obtained by saving the following as conf_ph.py

"""Configures a simple drudge for particle-hole model."""

from dummy_spark import SparkContext
from drudge import PartHoleDrudge

ctx = SparkContext()
dr = PartHoleDrudge(ctx)
dr.full_simplify = False

DRUDGE = dr

and then running the ccd_adv.drs found in the tutorial via

python -m drudge conf_ph.py ccd_adv.drs