| class ActivityInsightsJob(activityReader: SparkReader, | |
| analyticsReader: SparkReader, | |
| insightsWriter: SparkWriter | |
| )(implicit val sparkSession: SparkSession) extends SparkTask { | |
| def run(): Unit = { | |
| val metricsDF = analyticsReader.read(Some(AnalyticsSchema)) | |
| .transform(Events.isActivityImpression) | |
| .transform(Events.isActivityView) | |
| .transform(Events.isBookmarked) |
| export SPARK_OPTS='--packages org.apache.hadoop:hadoop-aws:2.8.5 --conf spark.hadoop.fs.s3a.aws.credentials.provider=org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider' | |
| pyspark |
Install Spark + Toree + Jupyter
pip install toree
jupyter toree install --spark_home=${SPARK_HOME} --sys-prefix
jupyter kernelspec list
Available kernels:
apache_toree_scala /Users/afranzi/.virtualenvs/mlflow/share/jupyter/kernels/apache_toree_scala
python3 /Users/afranzi/.virtualenvs/mlflow/share/jupyter/kernels/python3
Setup Environmnet variables
export SPARK_VERSION=2.4.0
export SPARK_PACKAGE=spark-${SPARK_VERSION}-bin-hadoop2.7
export SPARK_HOME=$HOME/spark-${SPARK_VERSION}
export PATH=$SPARK_HOME/bin:$PATH
export PYSPARK_DRIVER_PYTHON=jupyter
export PYSPARK_DRIVER_PYTHON_OPTS="notebook --notebook-dir=${HOME}/Projects/notebooks"
| CFLAGS="-I$(xcrun --show-sdk-path)/usr/include" pyenv install 3.6.7 | |
| pyenv global 3.6.7 | |
| mkvirtualenv mlflow | |
| workon mlflow |
Start by downloading the latest stable Apache Spark (current 2.4.3).
cd ~/Downloads/
tar -xzf spark-2.4.3-bin-hadoop2.7.tgz
mv ~/Downloads/spark-2.4.3-bin-hadoop2.7 ~/
ln -s ~/spark-2.4.3-bin-hadoop2.7 ~/spark̀
Install PySpark and Jupyter in our virtualEnv
| val schemasReferenced: Seq[SchemaReferenced] = Seq( | |
| SchemaReferenced("#", "/schemas/events/base-event/1.json"), | |
| SchemaReferenced("#", "/schemas/events/base-device-event/1.json"), | |
| SchemaReferenced("#", "/schemas/events/device-sensor-event/1.json"), | |
| SchemaReferenced("#/data", "/schemas/objects/sensors/SensorWifi/1.json"), | |
| SchemaReferenced("#/data/scan/[0]", "/schemas/objects/sensors/WifiConnection/1.json"), | |
| SchemaReferenced("#/data/scan/[1]", "/schemas/objects/sensors/WifiConnection/1.json"), | |
| SchemaReferenced("#/device", "/schemas/objects/Device/1.json"), | |
| SchemaReferenced("#/product", "/schemas/objects/Product/3.json"), | |
| SchemaReferenced("#/user", "/schemas/objects/User/2.json") |
| def evolve(event: ValidationResult[JsonNode]): EvolutionResult[JsonNode] = { | |
| val schemasReferenced: Seq[SchemaReferenced] = event.schemasReferenced | |
| val json = event.event | |
| val schemasToEvolve = schemasReferenced | |
| .filter { case SchemaReferenced(_, schemaRef) => hasEvolution(schemaRef) } | |
| val eventEvolved = schemasToEvolve | |
| .foldLeft(json) { | |
| case (jsonEvent: JsonNode, SchemaReferenced(location, schemaRef)) => |
| { | |
| "user": { | |
| "id": .userId | |
| }, | |
| "device": { | |
| "id": "undefined", | |
| "platform": if(.p == "a") "Android" else "iPhone" | |
| }, | |
| "product": { | |
| "id": "remix", |
| import SchemaValidationListener._ | |
| import org.everit.json.schema.Schema | |
| import org.everit.json.schema.event.{CombinedSchemaMatchEvent, SchemaReferencedEvent, ValidationListener} | |
| import scala.collection.mutable.ListBuffer | |
| import scala.util.matching.Regex | |
| import scala.collection.JavaConverters._ | |
| object SchemaValidationListener { | |