该问题针对的是熟悉py4j的人员-可以帮助解决酸洗错误。我正在尝试向pyspark
PythonMLLibAPI添加一个方法，该方法接受namedtuple的RDD，进行一些工作，并以RDD的形式返回结果。

该方法是在PYthonMLLibAPI.trainALSModel（）方法之后建模的，其类似的 现有 相关部分为：

  def trainALSModel(
    ratingsJRDD: JavaRDD[Rating],.. )

用于建模新代码的 现有 python Rating类为：

class Rating(namedtuple("Rating",["user","product","rating"])):
    def __reduce__(self):
        return Rating,(int(self.user),int(self.product),float(self.rating))

这是尝试，所以这里是相关的类：

新的 python类pyspark.mllib.clustering.MatrixEntry：

from collections import namedtuple
class MatrixEntry(namedtuple("MatrixEntry",["x","y","weight"])):
    def __reduce__(self):
        return MatrixEntry,(long(self.x),long(self.y),float(self.weight))

___PythonMLLibAPI中的_ 新 方法foob​​arRDD：

  def foobarRdd(
    data: JavaRDD[MatrixEntry]): RDD[FooBarResult] = {
    val rdd = data.rdd.map { d => FooBarResult(d.i,d.j,d.value,d.i * 100 + d.j * 10 + d.value)}
    rdd
  }

现在让我们尝试一下：

from pyspark.mllib.clustering import MatrixEntry

def convert_to_MatrixEntry(tuple):
  return MatrixEntry(*tuple)

from pyspark.mllib.clustering import *
pic = PowerIterationClusteringModel(2)
tups = [(1,2,3),(4,5,6),(12,13,14),(15,7,8),(16,17,16.5)]
trdd = sc.parallelize(map(convert_to_MatrixEntry,tups))

# print out the RDD on python side just for validation
print "%s" %(repr(trdd.collect()))

from pyspark.mllib.common import callMLlibFunc
pic = callMLlibFunc("foobar",trdd)

结果的相关部分：

[(1,2)=3.0,5)=6.0,13)=14.0,7)=8.0,17)=16.5]

这表明输入rdd为“整个”。但是，酸洗不开心：

5/04/27 21:15:44 ERROR Executor: Exception in task 6.0 in stage 1.0 (TID 14)
net.razorvine.pickle.PickleException: expected zero arguments for construction of ClassDict
(for pyspark.mllib.clustering.MatrixEntry)
    at net.razorvine.pickle.objects.ClassDictConstructor.construct(ClassDictConstructor.java:23)
    at net.razorvine.pickle.Unpickler.load_reduce(Unpickler.java:617)
    at net.razorvine.pickle.Unpickler.dispatch(Unpickler.java:170)
    at net.razorvine.pickle.Unpickler.load(Unpickler.java:84)
    at net.razorvine.pickle.Unpickler.loads(Unpickler.java:97)
    at org.apache.spark.mllib.api.python.SerDe$$anonfun$pythonToJava$1$$anonfun$apply$1.apply(PythonMLLibAPI.scala:1167)
    at org.apache.spark.mllib.api.python.SerDe$$anonfun$pythonToJava$1$$anonfun$apply$1.apply(PythonMLLibAPI.scala:1166)
    at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
    at scala.collection.Iterator$class.foreach(Iterator.scala:727)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
    at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
    at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
    at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
    at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
    at scala.collection.AbstractIterator.to(Iterator.scala:1157)
    at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
    at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
    at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
    at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
    at org.apache.spark.rdd.RDD$$anonfun$17.apply(RDD.scala:819)
    at org.apache.spark.rdd.RDD$$anonfun$17.apply(RDD.scala:819)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1523)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1523)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
    at org.apache.spark.scheduler.Task.run(Task.scala:64)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:212)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:724)

下面是python调用堆栈跟踪的视觉效果：

sqlDataReader reader = mocks.CreateMock<sqlDataReader>();
 Expect.Call(reader.Read()).Return(true).Repeat.Times(1);
 Expect.Call(reader.Read()).Return(false);
 Expect.Call(reader.HasRows).Return(true);
 Expect.Call(reader.dispose);
 Expect.Call(reader["City"]).Return("Boise");
 Expect.Call(reader["State"]).Return("State");
 Expect.Call(reader["LAT"]).Return(100);
 Expect.Call(reader["LON"]).Return(-100);
 mocks.ReplayAll();

using (reader)
        {
            if (reader.HasRows)
            {
                while (reader.Read())
                {
                    CityState myCity = new CityState
                       {
                           City = reader["City"].ToString(),State = reader["State"].ToString(),Lat = Convert.ToInt32(reader["LAT"]),Lon = Convert.ToInt32(reader["LON"])
                       };
                    myCities.Add(myCity);
                }                    
            }               
        }

解决方法

// Arrange
IDataReader reader = MockRepository.GenerateStub<IDataReader>();
reader.Stub(x => x.Read()).Repeat.Once().Return(true);
reader.Stub(x => x.Read()).Return(false);
reader.Stub(x => x["City"]).Return("Boise");
reader.Stub(x => x["State"]).Return("State");
reader.Stub(x => x["LAT"]).Return(100);
reader.Stub(x => x["LON"]).Return(-100);

// Act
var myCities = new List<CityState>();
using (reader)
{
    while (reader.Read())
    {
        CityState myCity = new CityState
        {
            City = reader["City"].ToString(),Lon = Convert.ToInt32(reader["LON"])
        };
        myCities.Add(myCity);
    }
}

// Assert
Assert.AreEqual(1,myCities.Count);
Assert.AreEqual("Boise",myCities[0].City);
...

我正在获得以下堆栈跟踪…

java.lang.ClassNotFoundException: org.hibernate.service.jndi.JndiException
    java.net.URLClassLoader$1.run(Unknown Source)
    java.net.URLClassLoader$1.run(Unknown Source)
    java.security.AccessController.doPrivileged(Native Method)
    java.net.URLClassLoader.findClass(Unknown Source)
    org.springframework.boot.loader.LaunchedURLClassLoader.findClass(LaunchedURLClassLoader.java:148)
    java.lang.ClassLoader.loadClass(Unknown Source)
    org.springframework.boot.loader.LaunchedURLClassLoader.doLoadClass(LaunchedURLClassLoader.java:131)
    org.springframework.boot.loader.LaunchedURLClassLoader.loadClass(LaunchedURLClassLoader.java:103)
    java.lang.ClassLoader.loadClass(Unknown Source)
    java.lang.Class.forName0(Native Method)
    java.lang.Class.forName(Unknown Source)
    org.jboss.logging.Logger$1.run(Logger.java:2252)
    java.security.AccessController.doPrivileged(Native Method)
    org.jboss.logging.Logger.getMessageLogger(Logger.java:2227)
    org.jboss.logging.Logger.getMessageLogger(Logger.java:2214)
    org.hibernate.ejb.Ejb3Configuration.<clinit>(Ejb3Configuration.java:143)
    org.hibernate.ejb.HibernatePersistence.createContainerEntityManagerFactory(HibernatePersistence.java:74)
    org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean.createNativeEntityManagerFactory(LocalContainerEntityManagerFactoryBean.java:318)
    org.springframework.orm.jpa.AbstractEntityManagerFactoryBean.afterPropertiesSet(AbstractEntityManagerFactoryBean.java:317)
    org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeInitMethods(AbstractAutowireCapableBeanFactory.java:1607)
    org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1544)
    org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:539)
    org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:475)
    org.springframework.beans.factory.support.AbstractBeanFactory$1.getObject(AbstractBeanFactory.java:300)
    org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:228)
    org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:296)
    org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:195)
    org.springframework.context.support.AbstractApplicationContext.getBean(AbstractApplicationContext.java:973)
    org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:750)
    org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:482)
    org.springframework.context.support.ClassPathXmlApplicationContext.<init>(ClassPathXmlApplicationContext.java:139)
    org.springframework.context.support.ClassPathXmlApplicationContext.<init>(ClassPathXmlApplicationContext.java:83)
    com.mydomain.GetController.getUser(GetController.java:20)
    sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
    sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    java.lang.reflect.Method.invoke(Unknown Source)
    org.springframework.web.method.support.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:214)
    org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:132)
    org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:104)
    org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandleMethod(RequestMappingHandlerAdapter.java:748)
    org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:689)
    org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:83)
    org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:947)
    org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:878)
    org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:946)
    org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:837)
    javax.servlet.http.HttpServlet.service(HttpServlet.java:621)
    org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:822)
    javax.servlet.http.HttpServlet.service(HttpServlet.java:728)
    org.springframework.web.filter.HiddenHttpMethodFilter.doFilterInternal(HiddenHttpMethodFilter.java:77)
    org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:108)

当我的Spring控制器处理连接时：

@RequestMapping(value = "/user",method = RequestMethod.GET)
    public @ResponseBody User getUser(@RequestParam(value="id",required=true) int id) {
        User user = null;

        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("spring-config.xml");
        UserService userService = (UserService) context.getBean("userService");

        user = userService.getUser(id);
        context.close();

        return user;
    }

它在userService上执行getUser：

import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class UserService {

    private UserDAO userDao;

    public UserDAO getUserDao() {
        return userDao;
    }

    @Autowired
    public void setUserDao(UserDAO userDao) {
        this.userDao = userDao;
    }

    public void addUser(User user) {
        getUserDao().insert(user);
    }

    public User getUser(int id) {
        return getUserDao().load(id);
    }

    public List<User> fetchAllPersons() {
        return getUserDao().selectAll();
    }
}

使用userDao查找实体：

import java.util.List;

import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;
import javax.persistence.Query;

import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

@Repository("userDao")
@Transactional(propagation = Propagation.REQUIRED)
public class UserDAO {

    private static final String SELECT_QUERY = "select u from users";

    @Qualifier(value = "entityManagerFactory")
    private EntityManager entityManager;

    public EntityManager getEntityManager() {
        return entityManager;
    }

    public void setEntityManager(EntityManager entityManager) {
        this.entityManager = entityManager;
    }

    public void insert(User user) {
        entityManager.persist(user);
    }

    public User load(int id) {
        return entityManager.find(User.class,id);
    }

    public List<User> selectAll() {
        Query query = entityManager.createQuery(SELECT_QUERY);
        @SuppressWarnings("unchecked")
        List<User> users = (List<User>) query.getResultList();
        return users;
    }

}

我在/ src / main / resources中具有以下spring-config.xml：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop"
    xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
           http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
           http://www.springframework.org/schema/aop
           http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
           http://www.springframework.org/schema/context
           http://www.springframework.org/schema/context/spring-context-3.0.xsd
           http://www.springframework.org/schema/tx  
           http://www.springframework.org/schema/tx/spring-tx.xsd">

    <context:component-scan base-package="com.mydomain.orm" />

    <bean id="entityManagerFactory">
        <property name="persistenceXmlLocation" value="classpath:/persistence.xml" />
        <property name="persistenceUnitName" value="userPersistenceUnit" />
        <property name="dataSource" ref="dataSource" />
        <property name="jpaVendorAdapter" ref="jpaVendorAdapter" />
        <property name="jpaDialect" ref="jpaDialect" />
    </bean>

    <bean id="jpaVendorAdapter">
        <property name="database" value="HSQL" />
        <property name="databasePlatform" value="org.hibernate.dialect.HSQLDialect" />
    </bean>

    <bean id="jpaDialect"/>

    <bean id="transactionManager">
        <property name="entityManagerFactory" ref="entityManagerFactory" />
        <property name="dataSource" ref="dataSource" />
        <property name="jpaDialect" ref="jpaDialect" />
    </bean>

    <tx:annotation-driven transaction-manager="transactionManager" />

    <bean id="dataSource">
        <property name="driverClassName" value="org.postgresql.Driver" />
        <property name="url" value="jdbc:postgresql://localhost:5432/mydb" />
        <property name="username" value="my_user" />
        <property name="password" value="" />
    </bean>
</beans>

最后但并非最不重要的是以下/src/main/resources/persistence.xml：

<?xml version="1.0" encoding="UTF-8"?>
<persistence xmlns="http://java.sun.com/xml/ns/persistence"
    version="1.0">
    <persistence-unit name="userPersistenceUnit" transaction-type="RESOURCE_LOCAL" >
    <class>com.mydomain.orm.User</class>
</persistence-unit>
</persistence>

我需要怎么做才能解决此错误？

不过，老实说，解决此问题后，我可能只会得到JndiException。您认为是什么原因造成的？