Author: zly Date: Mon Mar 20 21:59:36 2017 New Revision: 1787852 URL: http://svn.apache.org/viewvc?rev=1787852&view=rev Log: PIG-5187:UdfDistributedCache_1 is failing with spark exec type (Nandor via Liyun)
Added: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkPOUserFuncVisitor.java Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkLauncher.java Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkLauncher.java URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkLauncher.java?rev=1787852&r1=1787851&r2=1787852&view=diff ============================================================================== --- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkLauncher.java (original) +++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkLauncher.java Mon Mar 20 21:59:36 2017 @@ -232,7 +232,7 @@ public class SparkLauncher extends Launc } private void uploadResources(SparkOperPlan sparkPlan) throws IOException { - addFilesToSparkJob(); + addFilesToSparkJob(sparkPlan); addJarsToSparkJob(sparkPlan); } @@ -350,7 +350,7 @@ public class SparkLauncher extends Launc } } - private void addFilesToSparkJob() throws IOException { + private void addFilesToSparkJob(SparkOperPlan sparkPlan) throws IOException { LOG.info("add files Spark Job"); String shipFiles = pigContext.getProperties().getProperty( "pig.streaming.ship.files"); @@ -358,12 +358,22 @@ public class SparkLauncher extends Launc String cacheFiles = pigContext.getProperties().getProperty( "pig.streaming.cache.files"); cacheFiles(cacheFiles); + addUdfResourcesToSparkJob(sparkPlan); } + private void addUdfResourcesToSparkJob(SparkOperPlan sparkPlan) throws IOException { + SparkPOUserFuncVisitor sparkPOUserFuncVisitor = new SparkPOUserFuncVisitor(sparkPlan); + sparkPOUserFuncVisitor.visit(); + Joiner joiner = Joiner.on(","); + String shipFiles = joiner.join(sparkPOUserFuncVisitor.getShipFiles()); + shipFiles(shipFiles); + String cacheFiles = joiner.join(sparkPOUserFuncVisitor.getCacheFiles()); + cacheFiles(cacheFiles); + } private void shipFiles(String shipFiles) throws IOException { - if (shipFiles != null) { + if (shipFiles != null && !shipFiles.isEmpty()) { for (String file : shipFiles.split(",")) { File shipFile = new File(file.trim()); if (shipFile.exists()) { @@ -376,7 +386,7 @@ public class SparkLauncher extends Launc } private void cacheFiles(String cacheFiles) throws IOException { - if (cacheFiles != null) { + if (cacheFiles != null && !cacheFiles.isEmpty()) { File tmpFolder = Files.createTempDirectory("cache").toFile(); tmpFolder.deleteOnExit(); for (String file : cacheFiles.split(",")) { Added: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkPOUserFuncVisitor.java URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkPOUserFuncVisitor.java?rev=1787852&view=auto ============================================================================== --- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkPOUserFuncVisitor.java (added) +++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkPOUserFuncVisitor.java Mon Mar 20 21:59:36 2017 @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.pig.backend.hadoop.executionengine.spark; + +import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.UdfCacheShipFilesVisitor; +import org.apache.pig.backend.hadoop.executionengine.spark.plan.SparkOpPlanVisitor; +import org.apache.pig.backend.hadoop.executionengine.spark.plan.SparkOperPlan; +import org.apache.pig.backend.hadoop.executionengine.spark.plan.SparkOperator; +import org.apache.pig.impl.plan.DepthFirstWalker; +import org.apache.pig.impl.plan.VisitorException; + +import java.util.HashSet; +import java.util.Set; + +public class SparkPOUserFuncVisitor extends SparkOpPlanVisitor { + private Set<String> cacheFiles = new HashSet<>(); + private Set<String> shipFiles = new HashSet<>(); + + public SparkPOUserFuncVisitor(SparkOperPlan plan) { + super(plan, new DepthFirstWalker<>(plan)); + } + + @Override + public void visitSparkOp(SparkOperator sparkOperator) throws VisitorException { + if(!sparkOperator.physicalPlan.isEmpty()) { + UdfCacheShipFilesVisitor udfCacheFileVisitor = new UdfCacheShipFilesVisitor(sparkOperator.physicalPlan); + udfCacheFileVisitor.visit(); + cacheFiles.addAll(udfCacheFileVisitor.getCacheFiles()); + shipFiles.addAll(udfCacheFileVisitor.getShipFiles()); + } + } + + public Set<String> getCacheFiles() { + return cacheFiles; + } + + public Set<String> getShipFiles() { + return shipFiles; + } +}