diff --git a/spark/src/cgcloud/spark/__init__.py b/spark/src/cgcloud/spark/__init__.py index dc88eae..bfc4d4f 100644 --- a/spark/src/cgcloud/spark/__init__.py +++ b/spark/src/cgcloud/spark/__init__.py @@ -1,5 +1,5 @@ def roles( ): - from cgcloud.spark.spark_box import SparkBox, SparkSlave, SparkMaster + from cgcloud.spark.spark_box import SparkBox, Spark2Box, SparkSlave, SparkMaster return sorted( locals( ).values( ), key=lambda cls: cls.__name__ ) diff --git a/spark/src/cgcloud/spark/spark_box.py b/spark/src/cgcloud/spark/spark_box.py index 4c6c29b..ae3118e 100644 --- a/spark/src/cgcloud/spark/spark_box.py +++ b/spark/src/cgcloud/spark/spark_box.py @@ -25,10 +25,6 @@ persistent_dir = '/mnt/persistent' var_dir = '/var/lib/sparkbox' hdfs_replication = 1 -hadoop_version = '2.6.0' -spark_version = '1.6.2' -# The major version of Hadoop that the Spark binaries were built against -spark_hadoop_version = '2.6' Service = namedtuple( 'Service', [ 'init_name', @@ -82,6 +78,11 @@ class SparkBox( ApacheSoftwareBox, typically seconds after the reservation has been submitted, the slaves can be started up. """ + hadoop_version = '2.6.0' + spark_version = '1.6.2' + # The major version of Hadoop that the Spark binaries were built against + spark_hadoop_version = '2.6' + @classmethod def get_role_options( cls ): return super( SparkBox, cls ).get_role_options( ) + [ @@ -107,7 +108,11 @@ def _populate_security_group( self, group_id ): dict( ip_protocol='tcp', from_port=0, to_port=65535, src_security_group_group_id=group_id ), dict( ip_protocol='udp', from_port=0, to_port=65535, - src_security_group_group_id=group_id ) ] + src_security_group_group_id=group_id ), + dict( ip_protocol='tcp', from_port=8080, to_port=8081, + cidr_ip='0.0.0.0/0' ), + dict( ip_protocol='tcp', from_port=4040, to_port=4045, + cidr_ip='0.0.0.0/0' )] def _get_iam_ec2_role( self ): iam_role_name, policies = super( SparkBox, self )._get_iam_ec2_role( ) @@ -189,7 +194,7 @@ def __ec2_keypair_name( self, ctx ): @fabric_task def __install_hadoop( self ): # Download and extract Hadoop - path = fmt( 'hadoop/common/hadoop-{hadoop_version}/hadoop-{hadoop_version}.tar.gz' ) + path = fmt( 'hadoop/common/hadoop-{self.hadoop_version}/hadoop-{self.hadoop_version}.tar.gz' ) self._install_apache_package( path, install_dir ) # Add environment variables to hadoop_env.sh @@ -258,7 +263,7 @@ def __to_hadoop_xml_config( properties ): @fabric_task def __install_spark( self ): # Download and extract Spark - path = fmt( 'spark/spark-{spark_version}/spark-{spark_version}-bin-hadoop{spark_hadoop_version}.tgz' ) + path = fmt( 'spark/spark-{self.spark_version}/spark-{self.spark_version}-bin-hadoop{self.spark_hadoop_version}.tgz' ) self._install_apache_package( path, install_dir ) spark_dir = var_dir + "/spark" @@ -271,8 +276,9 @@ def __install_spark( self ): SPARK_WORKER_DIR=self._lazy_mkdir( spark_dir, "work" ), SPARK_LOCAL_DIRS=self._lazy_mkdir( spark_dir, "local" ), JAVA_HOME='/usr/lib/jvm/java-8-oracle', - SPARK_MASTER_IP='spark-master', - HADOOP_CONF_DIR=fmt( "{install_dir}/hadoop/etc/hadoop" ) ) + SPARK_MASTER_HOST='spark-master', + HADOOP_CONF_DIR=fmt( "{install_dir}/hadoop/etc/hadoop" ), + SPARK_PUBLIC_DNS="$(curl -s http://169.254.169.254/latest/meta-data/public-hostname)" ) with remote_open( spark_env_sh_path, use_sudo=True ) as spark_env_sh: spark_env_sh.write( '\n' ) for name, value in spark_env.iteritems( ): @@ -430,6 +436,18 @@ def __setup_path( self ): # Spark's and Hadoop's sbin f.write( fmt( 'PATH="$PATH:{install_dir}/{package}/bin"\n' ) ) +class Spark2Box( SparkBox ): + """ + A node in a Spark v2.x cluster; used only to create an image for master and worker boxes + """ + + hadoop_version = '2.7.3' + spark_version = '2.1.0' + # The major version of Hadoop that the Spark binaries were built against + spark_hadoop_version = '2.7' + + def __init__( self, ctx ): + super( Spark2Box, self ).__init__( ctx ) class SparkMaster( SparkBox, ClusterLeader ): """ diff --git a/spark/src/cgcloud/spark/test/test_spark.py b/spark/src/cgcloud/spark/test/test_spark.py index e292c51..96adc48 100644 --- a/spark/src/cgcloud/spark/test/test_spark.py +++ b/spark/src/cgcloud/spark/test/test_spark.py @@ -4,39 +4,49 @@ import time import logging import unittest +from abc import ABCMeta, abstractmethod from cgcloud.core.test import CoreTestCase -from cgcloud.spark.spark_box import install_dir, SparkBox, SparkMaster, SparkSlave +from cgcloud.spark.spark_box import install_dir, SparkBox, Spark2Box, SparkMaster, SparkSlave log = logging.getLogger( __name__ ) master = SparkMaster.role( ) slave = SparkSlave.role( ) -node = SparkBox.role( ) num_slaves = 2 -class SparkClusterTests( CoreTestCase ): +class BaseSparkClusterTests( CoreTestCase ): """ Covers the creation of a Spark cluster from scratch and running a simple Spark job on it. Also covers persistant HDFS between two cluster incarnations. """ + __metaclass__ = ABCMeta + + node = NotImplemented cleanup = True create_image = True + @classmethod + @abstractmethod + def initNode( cls ): + raise NotImplementedError("Abstract method") + @classmethod def setUpClass( cls ): + cls.initNode os.environ[ 'CGCLOUD_PLUGINS' ] = 'cgcloud.spark' - super( SparkClusterTests, cls ).setUpClass( ) + super( BaseSparkClusterTests, cls ).setUpClass( ) if cls.create_image: - cls._cgcloud( 'create', node, '-IT' ) + cls._cgcloud( 'create', cls.node, '-IT' ) @classmethod def tearDownClass( cls ): if cls.cleanup and cls.create_image: - cls._cgcloud( 'delete-image', node ) - super( SparkClusterTests, cls ).tearDownClass( ) + cls._cgcloud( 'delete-image', cls.node ) + super( BaseSparkClusterTests, cls ).tearDownClass( ) + def test_wordcount( self ): self._create_cluster( ) @@ -131,3 +141,24 @@ def word_count( ): def _delete_volumes( self ): pass + +class SparkClusterTests( BaseSparkClusterTests ): + """ + Covers the creation of a Spark v1.x cluster from scratch and running a simple Spark job on it. + Also covers persistant HDFS between two cluster incarnations. + """ + + @classmethod + def initNode( cls ): + cls.node = SparkBox.role( ) + + +class Spark2ClusterTests( BaseSparkClusterTests ): + """ + Covers the creation of a Spark v2.x cluster from scratch and running a simple Spark job on it. + Also covers persistant HDFS between two cluster incarnations. + """ + + @classmethod + def initNode( cls ): + cls.node = Spark2Box.role( )