[PIG-4059] Pig on Spark

• Type: New Feature
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: spark-branch, 0.17.0
• Component/s: spark
• Labels: spork
• Hadoop Flags: Reviewed
• Assignee: Praveen Rachabattuni
• Reporter: Rohini Palaniswamy
• Created: 17/Jul/14 14:54
• Updated: 21/Jun/17 09:15
• Resolved: 29/May/17 21:05

Description

Setting up your development environment:
0. download spark release package(currently pig on spark only support spark 1.6).
1. Check out Pig Spark branch.

2. Build Pig by running "ant jar" and "ant -Dhadoopversion=23 jar" for hadoop-2.x versions

3. Configure these environmental variables:
export HADOOP_USER_CLASSPATH_FIRST="true"
Now we support “local” and "yarn-client" mode, you can export system variable “SPARK_MASTER” like:
export SPARK_MASTER=local or export SPARK_MASTER="yarn-client"

4. In local mode: ./pig -x spark_local xxx.pig
In yarn-client mode:
export SPARK_HOME=xx;
export SPARK_JAR=hdfs://example.com:8020/xxxx (the hdfs location where you upload the spark-assembly*.jar)
./pig -x spark xxx.pig

Issue Links

• is related to: PIG-3446 Umbrella jira for Pig on Tez
• is related to: PIG-4266 Umbrella jira for unit tests for Spark
• supercedes: PIG-1734 Pig needs a more efficient DAG execution

Sub-Tasks

• Initial implementation of Pig on Spark: Closed
• Initial implementation of unit tests for Pig on Spark: Closed
• Move to Spark 1.x: Closed
• e2e tests for Spark: Closed
• Fix classpath error when using pig command with Spark: Resolved
• Make collected group work with Spark: Closed
• Make cross join work with Spark: Resolved
• Implement replicated join in Spark engine: Closed
• Make skewed join work with Spark: Closed
• Make ruby udfs work with Spark: Closed
• Make merge join work with Spark engine: Resolved
• Make python udfs work with Spark: Closed
• Make merge-sparse join work with Spark: Closed
• Make stream work with Spark: Closed
• Copy spark dependencies to lib directory: Closed
• Make rank work with Spark: Closed
• UDFContext is not initialized in executors when running on Spark cluster: Closed
• Package pig along with dependencies into a fat jar while job submission to Spark cluster: Closed
• Avoid packaging spark specific jars into pig fat jar: Closed
• Add SparkPlan in spark package: Closed
• Add stats and error reporting for Spark: Closed
• Move to Spark 1.2: Closed
• Merge from trunk (1) [Spark Branch]: Closed
• Merge from trunk (2) [Spark Branch]: Closed
• Upgrade to Spark 1.3: Closed
• change from "SparkLauncher#physicalToRDD" to "SparkLauncher#sparkPlanToRDD" after using spark plan in SparkLauncher: Closed
• Implement MergeJoin (as regular join) for Spark engine: Closed
• implement visitSkewedJoin in SparkCompiler: Closed
• Fix the NPE of System.getenv("SPARK_MASTER") in SparkLauncher.java: Closed
• remove unnessary MR plan code generated in SparkLauncher.java: Resolved
• Make ship work with spark: Closed
• PackageConverter hanging in Spark: Patch Available
• StackOverflowError in LIMIT operation on Spark: Patch Available
• Error when there is a bag inside an RDD: Closed
• "pig.output.lazy" not works in spark mode: Closed
• e2e tests for Spark can not work in hadoop env: Closed
• SchemaTupleBackend error when working on a Spark 1.1.0 cluster: Open
• Order By error after Group By in Spark: Closed
• Limit after sort does not work in spark mode: Closed
• Sort the leaves by SparkOperator.operatorKey in SparkLauncher#sparkOperToRDD: Closed
• Remove redundant code, comments in SparkLauncher: Closed
• Add apache license header to all spark package source files: Closed
• Enable Secondary key sort feature in spark mode: Closed
• Remove unnecessary store and load when POSplit is encounted: Closed
• SparkOperator should correspond to complete Spark job: Closed
• Enable local mode tests for Spark engine: Closed
• Remove repetitive org.apache.pig.test.Util#isSparkExecType: Closed
• OutputConsumerIterator should flush buffered records: Resolved
• Set CROSS operation parallelism for Spark engine: Closed
• Fix POGlobalRearrangeSpark copy constructor for Spark engine: Closed
• Modify the test.output value from "no" to "yes" to show more error message: Closed
• Support custom MR partitioners for Spark engine: Closed
• Fix unit test failure in TestSecondarySortSpark: Closed
• Pass value to MR Partitioners in Spark engine: Open
• Use "cogroup" spark api to implement "groupby+secondarysort" case in GlobalRearrangeConverter.java: Closed
• Enable "TestPruneColumn" in spark mode: Closed
• Use newAPIHadoopRDD instead of newAPIHadoopFile: Closed
• Cleanup: Rename POConverter to RDDConverter: Closed
• Move tests under 'test-spark' target: Closed
• Fix unit test failure in TestCase: Closed
• Enable "TestMultiQueryLocal" in spark mode: Closed
• Enable "TestMultiQuery" in spark mode: Closed
• Fix unit test failures about TestFRJoinNullValue in spark mode: Closed
• Fix unit test failures about MergeJoinConverter in spark mode: Closed
• Enable "TestNullConstant" unit test in spark mode: Closed
• Implement Merge CoGroup for Spark engine: Closed
• Clean up: refactor the package import order in the files under pig/src/org/apache/pig/backend/hadoop/executionengine/spark according to certain rule: Closed
• fix a bug when coping Jar to SparkJob working directory: Closed
• Enable "TestDefaultDateTimeZone" unit tests in spark mode: Closed
• Enable "TestRank1","TestRank3" unit tests in spark mode: Closed
• Enable "TestOrcStorage“ unit test in spark mode: Closed
• Fix remaining unit test failures about "TestHBaseStorage" in spark mode: Closed
• Fix unit test failures about TestAssert: Closed
• Enable "TestLocationInPhysicalPlan" in spark mode: Closed
• Fix null keys join in SkewedJoin in spark mode: Closed
• Fix UT errors of TestPigRunner in Spark mode: Closed
• Cleanup: change the indent size of some files of pig on spark project from 2 to 4 space: Closed
• Enable Illustrate in spark: In Progress
• Skip TestCubeOperator.testIllustrate and TestMultiQueryLocal.testMultiQueryWithIllustrate: Closed
• Update hadoop version to enable Spark output statistics: Closed
• Fix records count issues in output statistics: Closed
• Support hadoop-like Counter using spark accumulator: Closed
• Support InputStats in spark mode: Closed
• Fix unit test failures in org.apache.pig.test.TestScriptLanguageJavaScript: Closed
• Add Spark Unit Tests for SparkPigStats: Open
• Fix UT failures in TestPigServerLocal: Closed
• Enable Pig on Spark to run on Yarn Client mode: Closed
• Operators with multiple predecessors fail under multiquery optimization: Closed
• Enable Pig on Spark to run on Yarn Cluster mode: Resolved
• Class conflicts: Kryo bundled in spark vs kryo bundled with pig: Closed
• Enable dynamic resource allocation/de-allocation on Yarn backends: Closed
• Support combine for spark mode: Closed
• Tests in TestCombiner fail due to missing leveldb dependency: Closed
• Spark related JARs are not included when importing project via IDE: Closed
• the value of $SPARK_DIST_CLASSPATH in pig file is invalid: Resolved
• Ensure spark can be run as PIG action in Oozie: Open
• Fix UT failures in TestScriptLanguage: Closed
• Ensure GroupBy is optimized for all algebraic Operations: Closed
• Refactor SparkLauncher for spark engine: Closed
• Enable "pig.disable.counter“ for spark engine: Closed
• the value BytesRead metric info always returns 0 even the length of input file is not 0 in spark engine: Resolved
• Implement to collect metric data like getSMMSpillCount() in SparkJobStats: Open
• Merge trunk[3] into spark branch: Closed
• Collected group doesn't work in some cases: Closed
• pig.noSplitCombination=true should always be set internally for a merge join: Closed
• Merge trunk[4] into spark branch: Closed
• Last record is missing in STREAM operator: Closed
• Need upgrade snappy-java.version to 1.1.1.3: Closed
• OutputConsumeIterator can't handle the last buffered tuples for some Operators: Closed
• Add PigSplit#getLocationInfo to fix the NPE found in log in spark mode: Resolved
• Implement FR join by broadcasting small rdd not making more copys of data: Closed
• Fix unit test failure after PIG-4771's patch was checked in: Closed
• The number of records of input file is calculated wrongly in spark mode in multiquery case: Closed
• Fail to use Javascript UDF in spark yarn client mode: Closed
• Commit changes from last round of review on rb: Closed
• Remove schema tuple reference overhead for replicate join hashmap in POFRJoinSpark: Open
• Upgrade spark to 2.0: Closed
• Replace IndexedKey with PigNullableWritable in spark branch: Resolved
• exclude jline in spark dependency: Closed
• Duplicate record key info in GlobalRearrangeConverter#ToGroupKeyValueFunction: Closed
• Investigate why there are duplicated A[3,4] inTestLocationInPhysicalPlan#test in spark mode: Open
• Fix TestPigRunner#simpleMultiQueryTest3 in spark mode for wrong inputStats: Open
• Specify the hdfs path directly to spark and avoid the unnecessary download and upload in SparkLauncher.java: Open
• Implement auto parallelism for pig on spark: Open