{"action":"created","issue":{"url":"https://api.github.com/repos/hankcs/HanLP/issues/588","repository_url":"https://api.github.com/repos/hankcs/HanLP","labels_url":"https://api.github.com/repos/hankcs/HanLP/issues/588/labels{/name}","comments_url":"https://api.github.com/repos/hankcs/HanLP/issues/588/comments","events_url":"https://api.github.com/repos/hankcs/HanLP/issues/588/events","html_url":"https://github.com/hankcs/HanLP/issues/588","id":244601728,"number":588,"title":"在Spark单机使用正常，map到节点上使用出现错误","user":{"login":"fishermanff","id":22761847,"avatar_url":"https://avatars2.githubusercontent.com/u/22761847?v=4","gravatar_id":"","url":"https://api.github.com/users/fishermanff","html_url":"https://github.com/fishermanff","followers_url":"https://api.github.com/users/fishermanff/followers","following_url":"https://api.github.com/users/fishermanff/following{/other_user}","gists_url":"https://api.github.com/users/fishermanff/gists{/gist_id}","starred_url":"https://api.github.com/users/fishermanff/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fishermanff/subscriptions","organizations_url":"https://api.github.com/users/fishermanff/orgs","repos_url":"https://api.github.com/users/fishermanff/repos","events_url":"https://api.github.com/users/fishermanff/events{/privacy}","received_events_url":"https://api.github.com/users/fishermanff/received_events","type":"User","site_admin":false},"labels":[{"id":136114038,"url":"https://api.github.com/repos/hankcs/HanLP/labels/%E6%B1%82%E5%8A%A9","name":"求助","color":"159818","default":false}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":12,"created_at":"2017-07-21T08:38:42Z","updated_at":"2018-03-22T03:00:31Z","closed_at":null,"author_association":"NONE","body":"<!--\r\n这是HanLP的issue模板，用于规范提问题的格式。本来并不打算用死板的格式限制大家，但issue区实在有点混乱。有时候说了半天才搞清楚原来对方用的是旧版、自己改了代码之类，浪费双方宝贵时间。所以这里用一个规范的模板统一一下，造成不便望海涵。除了注意事项外，其他部分可以自行根据实际情况做适量修改。\r\n-->\r\n\r\n## 注意事项\r\n请确认下列注意事项：\r\n\r\n* 我已仔细阅读下列文档，都没有找到答案：\r\n  - [首页文档](https://github.com/hankcs/HanLP)\r\n  - [wiki](https://github.com/hankcs/HanLP/wiki)\r\n  - [常见问题](https://github.com/hankcs/HanLP/wiki/FAQ)\r\n* 我已经通过[Google](https://www.google.com/#newwindow=1&q=HanLP)和[issue区检索功能](https://github.com/hankcs/HanLP/issues)搜索了我的问题，也没有找到答案。\r\n* 我明白开源社区是出于兴趣爱好聚集起来的自由社区，不承担任何责任或义务。我会礼貌发言，向每一个帮助我的人表示感谢。\r\n* [x] 我在此括号内输入x打钩，代表上述事项确认完毕。\r\n\r\n## 版本号\r\n<!-- 发行版请注明jar文件名去掉拓展名的部分；GitHub仓库版请注明master还是portable分支 -->\r\n\r\n当前最新版本号是：1.3.4\r\n我使用的版本是：1.3.4\r\n\r\n@hankcs 您好，按照您说的方法，修改配置\r\nroot=hdfs://localhost:9000/hanlpdata/\r\nIOAdapter=com.xxx.HDFSIOAdapter\r\n而且在scala重写了接口：\r\nclass HadoopFileIoAdapter extends IIOAdapter {\r\n    @Override\r\n    def open(path: String): java.io.InputStream = {\r\n        val conf: Configuration = new Configuration();\r\n        val fs: FileSystem = FileSystem.get(URI.create(path), conf);\r\n        fs.open(new Path(path));\r\n    }    \r\n\r\n    @Override\r\n    def create(path: String): java.io.OutputStream = {\r\n        val conf: Configuration = new Configuration();\r\n        val fs: FileSystem = FileSystem.get(URI.create(path), conf);\r\n        fs.create(new Path(path));\r\n    }\r\n}\r\nHanLP.Config.IOAdapter = new HadoopFileIoAdapter();\r\n\r\n也只能在单机spark下运行分词\r\nscala> System.out.println(HanLP.segment(\"你好，欢迎使用HanLP汉语处理包！\"));\r\n[你好/vl, ，/w, 欢迎/v, 使用/v, HanLP/nx, 汉语/gi, 处理/vn, 包/v, ！/w]\r\n\r\n\r\n分发到节点的时候就出错了，请问你们在集群上使用的时候是否遇到过这种问题？需要注意哪些坑呢？\r\nval a = sc.parallelize(Seq(\"中国的神威太湖之光计算机被用于天气预报\",\"制药研究和工业设计等领域。\"))\r\nval res = a.map(e=>{\r\n        HanLP.segment(e).toString\r\n    })\r\nres take 2 foreach println\r\n\r\nExit code: 255\r\nStack trace: ExitCodeException exitCode=255:\r\n\tat org.apache.hadoop.util.Shell.runCommand(Shell.java:578)\r\n\tat org.apache.hadoop.util.Shell.run(Shell.java:489)\r\n\tat org.apache.hadoop.util.Shell$ShellCommandExecutor.execute(Shell.java:755)\r\n\tat org.apache.hadoop.yarn.server.nodemanager.LinuxContainerExecutor.launchContainer(LinuxContainerExecutor.java:297)\r\n\tat org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch.call(ContainerLaunch.java:302)\r\n\tat org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch.call(ContainerLaunch.java:82)\r\n\tat java.util.concurrent.FutureTask.run(FutureTask.java:262)\r\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\r\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\r\n\tat java.lang.Thread.run(Thread.java:745)"},"comment":{"url":"https://api.github.com/repos/hankcs/HanLP/issues/comments/375163552","html_url":"https://github.com/hankcs/HanLP/issues/588#issuecomment-375163552","issue_url":"https://api.github.com/repos/hankcs/HanLP/issues/588","id":375163552,"user":{"login":"zongwu233","id":14315424,"avatar_url":"https://avatars1.githubusercontent.com/u/14315424?v=4","gravatar_id":"","url":"https://api.github.com/users/zongwu233","html_url":"https://github.com/zongwu233","followers_url":"https://api.github.com/users/zongwu233/followers","following_url":"https://api.github.com/users/zongwu233/following{/other_user}","gists_url":"https://api.github.com/users/zongwu233/gists{/gist_id}","starred_url":"https://api.github.com/users/zongwu233/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zongwu233/subscriptions","organizations_url":"https://api.github.com/users/zongwu233/orgs","repos_url":"https://api.github.com/users/zongwu233/repos","events_url":"https://api.github.com/users/zongwu233/events{/privacy}","received_events_url":"https://api.github.com/users/zongwu233/received_events","type":"User","site_admin":false},"created_at":"2018-03-22T03:00:31Z","updated_at":"2018-03-22T03:00:31Z","author_association":"NONE","body":"看了一下HanLP源码里关于初始化的部分，在第一次调用HanLP类的方法的时候，很多对象才被创建，data目录的资料才被加载。\r\n在spark上，其实只要避免在 RDD.map()方法之外创建HanLP相关类的对象，就避免了HanLP 相关类对象在 spark 节点之间传输，也就不需要改造目前的类为序列化类了。\r\n最简单的办法就像@cyq130  所说，在`RDD.foreachPartition` 方法里面再初始化HanLP即可。\r\n如果改造相关类为序列化类并在driver中就初始化HanLP，势必造成data的数据在driver加载（因为相关类被实例化），然后序列化给每个task。这样做不算是最佳方式。\r\n在`RDD.foreachPartition` 方法里初始化，让各个`Partition`自己加载数据，是在spark中访问外部资源的常用做法。比如建立与Hbase的connection。\r\n目前我的做法是：\r\n将data目录上传到spark集群，将该文件的HDFS URI 值写到 `hanlp.properties`的 `root `\r\nspark中调用HanLP的实例代码：\r\n```scala\r\nval message =\r\n    \"受一桩涉及 5000 万用户信息泄露事件的舆论打击，Facebook 的股价从 19 日开始经历了近乎断崖式的下跌，单日内最大跌幅达到 7.92% ... Facebook、阿里巴巴这两家公司的价值分析没有太多可比性，两家的业务交集微乎其微，且短期内影响公司股价与市值的因素十分琐碎，\"\r\n\r\n    val a = sparkContext.parallelize(message.split(\"，\"))\r\n    a.foreachPartition(it => {\r\n      HanLP.Config.IOAdapter = new HadoopFileIoAdapter\r\n      for (elem <- it) {\r\n        val result = HanLP.segment(elem)\r\n        loger.warn(\"start------\")\r\n        result.forEach(new Consumer[Term] {\r\n          override def accept(t: Term): Unit = {\r\n            loger.warn(\"segment:\" + t.word)\r\n          }\r\n        })\r\n\r\n      }\r\n\r\n    })\r\n```\r\n\r\n其中 HadoopFileIoAdapter  的实现：\r\n```scala\r\nclass HadoopFileIoAdapter extends IIOAdapter {\r\n\r\n  override def create(path: String): OutputStream = {\r\n    val conf: Configuration = new Configuration()\r\n    val fs: FileSystem = FileSystem.get(URI.create(path), conf)\r\n    fs.create(new Path(path))\r\n  }\r\n\r\n  override def open(path: String): InputStream = {\r\n    val conf: Configuration = new Configuration()\r\n    val fs: FileSystem = FileSystem.get(URI.create(path), conf)\r\n    fs.open(new Path(path))\r\n  }\r\n}\r\n```"}}
{"actor":{"display_login":"zongwu233"}}