{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Aerospike Spark Connector Tutorial for Scala\n", "\n", "## Tested with Spark connector 3.1.0, Java 8, Apache Spark 3.0.2, Python 3.7 and Scala 2.12.11 and [Spylon]( https://pypi.org/project/spylon-kernel/)\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "#### Please download the appropriate Aeropsike Connect for Spark from the [download page](https://enterprise.aerospike.com/enterprise/download/connectors/aerospike-spark/notes.html) \n", "Set `launcher.jars` with path to the downloaded binary" ] }, { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [], "source": [ "%%init_spark \n", "launcher.jars = [\"aerospike-spark-assembly-3.1.0.jar\"] \n", "launcher.master = \"local[*]\"" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "Intitializing Scala interpreter ..." ] }, "metadata": {}, "output_type": "display_data" }, { "data": { "text/plain": [ "Spark Web UI available at http://192.168.106.119:4040\n", "SparkContext available as 'sc' (version = 3.0.2, master = local[*], app id = local-1626732681010)\n", "SparkSession available as 'spark'\n" ] }, "metadata": {}, "output_type": "display_data" }, { "data": { "text/plain": [ "AS_HOST: String = 172.16.39.192:3000\n" ] }, "execution_count": 2, "metadata": {}, "output_type": "execute_result" } ], "source": [ "//Specify the Seed Host of the Aerospike Server\n", "val AS_HOST = \"172.16.39.192:3000\"" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "import scala.collection.mutable.ArrayBuffer\n", "import org.apache.spark.sql.Row\n", "import org.apache.spark.sql.types._\n", "import org.apache.spark.sql.functions._\n", "import org.apache.spark.sql.SaveMode\n", "import com.aerospike.spark.sql.AerospikeConnection\n", "import org.apache.spark.sql.SparkSession\n" ] }, "execution_count": 3, "metadata": {}, "output_type": "execute_result" } ], "source": [ "import scala.collection.mutable.ArrayBuffer\n", "import org.apache.spark.sql.Row\n", "import org.apache.spark.sql.types._\n", "import org.apache.spark.sql.functions._\n", "import org.apache.spark.sql.SaveMode\n", "import com.aerospike.spark.sql.AerospikeConnection\n", "import org.apache.spark.sql.SparkSession" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Schema in the Spark Connector\n", "\n", "- Aerospike is schemaless, however Spark adher to schema. After the schema is decided upon (either through inference or given), data within the bins must honor the types. \n", "\n", "- To infer the schema, the connector samples a set of records (configurable through `aerospike.schema.scan`) to decide the name of bins/columns and their types. This implies that the derived schema depends entirely upon sampled records. \n", "\n", "- **Note that `__key` was not part of provided schema. So how can one query using `__key`? We can just add `__key` in provided schema with appropriate type. Similarly we can add `__gen` or `__ttl` etc.** \n", " \n", " val schemaWithPK: StructType = new StructType(Array(\n", " StructField(\"__key\",IntegerType, nullable = false), \n", " StructField(\"id\", IntegerType, nullable = false),\n", " StructField(\"name\", StringType, nullable = false),\n", " StructField(\"age\", IntegerType, nullable = false),\n", " StructField(\"salary\",IntegerType, nullable = false)))\n", " \n", "- **We recommend that you provide schema for queries that involve [collection data types](https://docs.aerospike.com/docs/guide/cdt.html) such as lists, maps, and mixed types. Using schema inference for CDT may cause unexpected issues.** " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Create sample data and write it into Aerospike Database" ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+---+------+---+------+\n", "| id| name|age|salary|\n", "+---+------+---+------+\n", "| 1| name1| 1| 71425|\n", "| 2| name2| 2| 64969|\n", "| 3| name3| 3| 76504|\n", "| 4| name4| 4| 86652|\n", "| 5| name5| 5| 72894|\n", "| 6| name6| 6| 80305|\n", "| 7| name7| 7| 68467|\n", "| 8| name8| 8| 91715|\n", "| 9| name9| 9| 81021|\n", "| 10|name10| 10| 85134|\n", "+---+------+---+------+\n", "only showing top 10 rows\n", "\n" ] }, { "data": { "text/plain": [ "conf: org.apache.spark.SparkConf = org.apache.spark.SparkConf@43fd6f26\n", "spark2: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@32d3db7d\n", "num_records: Int = 1000\n", "rand: util.Random.type = scala.util.Random$@698c02f8\n", "schema: org.apache.spark.sql.types.StructType = StructType(StructField(id,IntegerType,false), StructField(name,StringType,false), StructField(age,IntegerType,false), StructField(salary,IntegerType,false))\n", "inputDF: org.apache.spark.sql.DataFrame = [id: int, name: string ... 2 more fields]\n" ] }, "execution_count": 4, "metadata": {}, "output_type": "execute_result" } ], "source": [ "//Create test data\n", "val conf = sc.getConf.clone();\n", "\n", "conf.set(\"aerospike.seedhost\" , AS_HOST)\n", "conf.set(\"aerospike.namespace\", \"test\")\n", "conf.set(\"aerospike.log.level\", \"info\")\n", "spark.close()\n", "val spark2= SparkSession.builder().config(conf).master(\"local[2]\").getOrCreate()\n", "\n", "val num_records=1000\n", "val rand = scala.util.Random\n", "\n", "\n", "val schema: StructType = new StructType(\n", " Array(\n", " StructField(\"id\", IntegerType, nullable = false),\n", " StructField(\"name\", StringType, nullable = false),\n", " StructField(\"age\", IntegerType, nullable = false),\n", " StructField(\"salary\",IntegerType, nullable = false)\n", " ))\n", "\n", "val inputDF = {\n", " val inputBuf= new ArrayBuffer[Row]()\n", " for ( i <- 1 to num_records){\n", " val name = \"name\" + i\n", " val age = i%100\n", " val salary = 50000 + rand.nextInt(50000)\n", " val id = i \n", " val r = Row(id, name, age,salary)\n", " inputBuf.append(r)\n", " }\n", " val inputRDD = spark2.sparkContext.parallelize(inputBuf.toSeq)\n", " spark2.createDataFrame(inputRDD,schema)\n", "}\n", "\n", "inputDF.show(10)\n", "\n", "//Write the Sample Data to Aerospike\n", "inputDF.write.mode(SaveMode.Overwrite) \n", ".format(\"aerospike\") //aerospike specific format\n", ".option(\"aerospike.writeset\", \"scala_input_data\") //write to this set\n", ".option(\"aerospike.updateByKey\", \"id\") //indicates which columns should be used for construction of primary key\n", ".option(\"aerospike.sendKey\", \"true\")\n", ".save()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Using Spark SQL syntax" ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+---+-------+---+------+\n", "| id| name|age|salary|\n", "+---+-------+---+------+\n", "|132|name132| 32| 54949|\n", "|647|name647| 47| 96580|\n", "| 45| name45| 45| 82480|\n", "|558|name558| 58| 91583|\n", "|608|name608| 8| 83286|\n", "|687|name687| 87| 96887|\n", "|372|name372| 72| 64562|\n", "|335|name335| 35| 59378|\n", "|911|name911| 11| 93950|\n", "|352|name352| 52| 57325|\n", "| 94| name94| 94| 70982|\n", "|890|name890| 90| 55053|\n", "|334|name334| 34| 88603|\n", "|907|name907| 7| 87233|\n", "|148|name148| 48| 62904|\n", "|315|name315| 15| 64200|\n", "|163|name163| 63| 62598|\n", "|882|name882| 82| 63215|\n", "|602|name602| 2| 73376|\n", "|673|name673| 73| 98636|\n", "+---+-------+---+------+\n", "only showing top 20 rows\n", "\n" ] }, { "data": { "text/plain": [ "insertDFWithSchema: org.apache.spark.sql.DataFrame = [id: int, name: string ... 2 more fields]\n", "sqlView: String = inserttable\n" ] }, "execution_count": 5, "metadata": {}, "output_type": "execute_result" } ], "source": [ "/*\n", "Aerospike DB needs a Primary key for record insertion. Hence, you must identify the primary key column \n", "using for example .option(“aerospike.updateByKey”, “id”), where “id” is the name of the column that you’d \n", "like to be the Primary key, while loading data from the DB.\n", "*/\n", "val insertDFWithSchema=spark2\n", ".sqlContext\n", ".read\n", ".format(\"aerospike\")\n", ".schema(schema)\n", ".option(\"aerospike.set\", \"scala_input_data\")\n", ".load()\n", "\n", "val sqlView=\"inserttable\"\n", "insertDFWithSchema.createOrReplaceTempView(sqlView)\n", "//\n", "//V2 datasource doesn't allow insert into a view. \n", "//\n", "\n", "spark2.sql(s\"select * from $sqlView\").show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Load data into a DataFrame without specifying any schema i.e. using connector schema inference" ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- __key: string (nullable = true)\n", " |-- __digest: binary (nullable = true)\n", " |-- __expiry: integer (nullable = false)\n", " |-- __generation: integer (nullable = false)\n", " |-- __ttl: integer (nullable = false)\n", " |-- age: long (nullable = true)\n", " |-- name: string (nullable = true)\n", " |-- salary: long (nullable = true)\n", " |-- id: long (nullable = true)\n", "\n" ] }, { "data": { "text/plain": [ "loadedDFWithoutSchema: org.apache.spark.sql.DataFrame = [__key: string, __digest: binary ... 7 more fields]\n" ] }, "execution_count": 6, "metadata": {}, "output_type": "execute_result" } ], "source": [ "// Create a Spark DataFrame by using the Connector Schema inference mechanism\n", "\n", "val loadedDFWithoutSchema=spark2\n", ".sqlContext\n", ".read\n", ".format(\"aerospike\")\n", ".option(\"aerospike.set\", \"scala_input_data\") //read the data from this set\n", ".load\n", "loadedDFWithoutSchema.printSchema()\n", "//Notice that schema of loaded data has some additional fields. \n", "// When connector infers schema, it also adds internal metadata." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Load data into a DataFrame with user specified schema " ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+---+-------+---+------+\n", "| id| name|age|salary|\n", "+---+-------+---+------+\n", "|132|name132| 32| 54949|\n", "|647|name647| 47| 96580|\n", "| 45| name45| 45| 82480|\n", "|558|name558| 58| 91583|\n", "|608|name608| 8| 83286|\n", "+---+-------+---+------+\n", "only showing top 5 rows\n", "\n" ] }, { "data": { "text/plain": [ "loadedDFWithSchema: org.apache.spark.sql.DataFrame = [id: int, name: string ... 2 more fields]\n" ] }, "execution_count": 7, "metadata": {}, "output_type": "execute_result" } ], "source": [ "//Data can be loaded with known schema as well.\n", "val loadedDFWithSchema=spark2\n", ".sqlContext\n", ".read\n", ".format(\"aerospike\")\n", ".schema(schema)\n", ".option(\"aerospike.set\", \"scala_input_data\").load\n", "loadedDFWithSchema.show(5)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Writing Sample Collection Data Types (CDT) data into Aerospike" ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- name: struct (nullable = true)\n", " | |-- first_name: string (nullable = true)\n", " | |-- aliases: array (nullable = true)\n", " | | |-- element: struct (containsNull = true)\n", " | | | |-- first_name: string (nullable = true)\n", " | | | |-- last_name: string (nullable = true)\n", " |-- SSN: string (nullable = true)\n", " |-- home_address: array (nullable = true)\n", " | |-- element: struct (containsNull = true)\n", " | | |-- zip: long (nullable = true)\n", " | | |-- street: struct (nullable = true)\n", " | | | |-- street_name: string (nullable = true)\n", " | | | |-- apt_number: integer (nullable = true)\n", " | | |-- city: string (nullable = true)\n", " |-- work_history: array (nullable = true)\n", " | |-- element: struct (containsNull = true)\n", " | | |-- company_name: string (nullable = true)\n", " | | |-- company_address: struct (nullable = true)\n", " | | | |-- zip: long (nullable = true)\n", " | | | |-- street: struct (nullable = true)\n", " | | | | |-- street_name: string (nullable = true)\n", " | | | | |-- apt_number: integer (nullable = true)\n", " | | | |-- city: string (nullable = true)\n", " | | |-- worked_from: string (nullable = true)\n", "\n" ] }, { "data": { "text/plain": [ "complex_data_json: String = resources/nested_data.json\n", "alias: org.apache.spark.sql.types.StructType = StructType(StructField(first_name,StringType,false), StructField(last_name,StringType,false))\n", "name: org.apache.spark.sql.types.StructType = StructType(StructField(first_name,StringType,false), StructField(aliases,ArrayType(StructType(StructField(first_name,StringType,false), StructField(last_name,StringType,false)),true),false))\n", "street_adress: org.apache.spark.sql.types.StructType = StructType(StructField(street_name,StringType,false), StructField(apt_number,IntegerType,false))\n", "address: org.apache.spark.sql.types.StructType = StructType(StructField(zip,LongType,false), StructField(street,StructType(StructField(street_name,StringType,false), StructField(apt_number,IntegerType,false)),fal...\n" ] }, "execution_count": 8, "metadata": {}, "output_type": "execute_result" } ], "source": [ "val complex_data_json=\"resources/nested_data.json\"\n", "val alias= StructType(List(\n", " StructField(\"first_name\",StringType, false),\n", " StructField(\"last_name\",StringType, false)))\n", "\n", " val name= StructType(List(\n", " StructField(\"first_name\",StringType, false),\n", " StructField(\"aliases\",ArrayType(alias), false )\n", " ))\n", "\n", " val street_adress= StructType(List(\n", " StructField(\"street_name\", StringType, false),\n", " StructField(\"apt_number\" , IntegerType, false)))\n", "\n", " val address = StructType( List(\n", " StructField (\"zip\" , LongType, false),\n", " StructField(\"street\", street_adress, false),\n", " StructField(\"city\", StringType, false)))\n", "\n", " val workHistory = StructType(List(\n", " StructField (\"company_name\" , StringType, false),\n", " StructField( \"company_address\" , address, false),\n", " StructField(\"worked_from\", StringType, false)))\n", "\n", " val person= StructType ( List(\n", " StructField(\"name\" , name, false, Metadata.empty),\n", " StructField(\"SSN\", StringType, false,Metadata.empty),\n", " StructField(\"home_address\", ArrayType(address), false),\n", " StructField(\"work_history\", ArrayType(workHistory), false)))\n", "\n", "val cmplx_data_with_schema=spark2.read.schema(person).json(complex_data_json)\n", "\n", "cmplx_data_with_schema.printSchema()\n", "cmplx_data_with_schema.write.mode(SaveMode.Overwrite) \n", ".format(\"aerospike\") //aerospike specific format\n", ".option(\"aerospike.seedhost\", AS_HOST) //db hostname, can be added multiple hosts, delimited with \":\"\n", ".option(\"aerospike.namespace\", \"test\") //use this namespace \n", ".option(\"aerospike.writeset\", \"scala_complex_input_data\") //write to this set\n", ".option(\"aerospike.updateByKey\", \"name.first_name\") //indicates which columns should be used for construction of primary key\n", ".save()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Load Complex Data Types (CDT) into a DataFrame with user specified schema" ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+--------------------+-----------+--------------------+--------------------+\n", "| name| SSN| home_address| work_history|\n", "+--------------------+-----------+--------------------+--------------------+\n", "|[Kurt, [[Jacob, R...|533-07-8760|[[37634, [Hammond...|[[Atkins Group, [...|\n", "|[Jamie, [[Patrici...|569-31-4715|[[53379, [James I...|[[Brown, Miller a...|\n", "+--------------------+-----------+--------------------+--------------------+\n", "only showing top 2 rows\n", "\n", "root\n", " |-- name: struct (nullable = false)\n", " | |-- first_name: string (nullable = false)\n", " | |-- aliases: array (nullable = false)\n", " | | |-- element: struct (containsNull = true)\n", " | | | |-- first_name: string (nullable = false)\n", " | | | |-- last_name: string (nullable = false)\n", " |-- SSN: string (nullable = false)\n", " |-- home_address: array (nullable = false)\n", " | |-- element: struct (containsNull = true)\n", " | | |-- zip: long (nullable = false)\n", " | | |-- street: struct (nullable = false)\n", " | | | |-- street_name: string (nullable = false)\n", " | | | |-- apt_number: integer (nullable = false)\n", " | | |-- city: string (nullable = false)\n", " |-- work_history: array (nullable = false)\n", " | |-- element: struct (containsNull = true)\n", " | | |-- company_name: string (nullable = false)\n", " | | |-- company_address: struct (nullable = false)\n", " | | | |-- zip: long (nullable = false)\n", " | | | |-- street: struct (nullable = false)\n", " | | | | |-- street_name: string (nullable = false)\n", " | | | | |-- apt_number: integer (nullable = false)\n", " | | | |-- city: string (nullable = false)\n", " | | |-- worked_from: string (nullable = false)\n", "\n" ] }, { "data": { "text/plain": [ "loadedComplexDFWithSchema: org.apache.spark.sql.DataFrame = [name: struct>>, SSN: string ... 2 more fields]\n", "res5: loadedComplexDFWithSchema.type = [name: struct>>, SSN: string ... 2 more fields]\n" ] }, "execution_count": 9, "metadata": {}, "output_type": "execute_result" } ], "source": [ "val loadedComplexDFWithSchema=spark2\n", ".sqlContext\n", ".read\n", ".format(\"aerospike\")\n", ".option(\"aerospike.set\", \"scala_complex_input_data\") //read the data from this set\n", ".schema(person)\n", ".load\n", "\n", "loadedComplexDFWithSchema.show(2)\n", "loadedComplexDFWithSchema.printSchema()\n", "loadedComplexDFWithSchema.cache()\n", "//Please note the difference in types of loaded data in both cases. With schema, we extactly infer complex types." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Quering Aerospike Data using SparkSQL\n", "\n", "### Things to keep in mind\n", " 1. Queries that involve Primary Key or Digest in the predicate trigger aerospike_batch_get()( https://www.aerospike.com/docs/client/c/usage/kvs/batch.html) and run extremely fast. For e.g. a query containing `__key` or `__digest` with, with no `OR` between two bins.\n", " 2. All other queries may entail a full scan of the Aerospike DB if they can’t be converted to Aerospike batchget. " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Queries that include Primary Key in the Predicate\n", "\n", "In case of batchget queries we can also apply filters upon metadata columns like `__gen` or `__ttl` etc. To do so, these columns should be exposed through schema (if schema provided). " ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----+--------------------+--------+------------+-----+---+-------+------+---+\n", "|__key| __digest|__expiry|__generation|__ttl|age| name|salary| id|\n", "+-----+--------------------+--------+------------+-----+---+-------+------+---+\n", "| 829|[C0 B6 C4 DE 68 D...| 0| 1| -1| 29|name829| 92238|829|\n", "+-----+--------------------+--------+------------+-----+---+-------+------+---+\n", "\n" ] }, { "data": { "text/plain": [ "batchGet1: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [__key: int, __digest: binary ... 7 more fields]\n" ] }, "execution_count": 10, "metadata": {}, "output_type": "execute_result" } ], "source": [ "val batchGet1= spark2.sqlContext\n", ".read\n", ".format(\"aerospike\")\n", ".option(\"aerospike.set\", \"scala_input_data\")\n", ".option(\"aerospike.keyType\", \"int\") //used to hint primary key(PK) type when schema is not provided.\n", ".load.where(\"__key = 829\")\n", "batchGet1.show()\n", "//Please be aware Aerospike database supports only equality test with PKs in primary key query. \n", "//So, a where clause with \"__key >10\", would result in scan query!" ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----+--------------------+--------+------------+-----+---+------+------+---+\n", "|__key| __digest|__expiry|__generation|__ttl|age| name|salary| id|\n", "+-----+--------------------+--------+------------+-----+---+------+------+---+\n", "| 1|[89 31 AB FE 54 D...| 0| 1| -1| 1| name1| 71425| 1|\n", "| 4|[93 F1 65 F0 E8 9...| 0| 1| -1| 4| name4| 86652| 4|\n", "| 3|[D4 A1 0B A5 12 0...| 0| 1| -1| 3| name3| 76504| 3|\n", "| 7|[30 94 D4 E7 9E 8...| 0| 1| -1| 7| name7| 68467| 7|\n", "| 5|[3E F5 94 A9 3A A...| 0| 1| -1| 5| name5| 72894| 5|\n", "| 14|[06 66 ED 38 08 F...| 0| 1| -1| 14|name14| 53533| 14|\n", "| 13|[EA 78 AB 39 FC C...| 0| 1| -1| 13|name13| 98475| 13|\n", "| 2|[41 DB A8 23 03 4...| 0| 1| -1| 2| name2| 64969| 2|\n", "| 8|[60 AB E7 17 C8 5...| 0| 1| -1| 8| name8| 91715| 8|\n", "| 9|[1B 6D CD D8 D2 5...| 0| 1| -1| 9| name9| 81021| 9|\n", "| 6|[C2 4D 37 CC 2B 2...| 0| 1| -1| 6| name6| 80305| 6|\n", "| 12|[F8 4E EC 27 8F 1...| 0| 1| -1| 12|name12| 80583| 12|\n", "| 10|[8D 0F 84 CD B0 7...| 0| 1| -1| 10|name10| 85134| 10|\n", "+-----+--------------------+--------+------------+-----+---+------+------+---+\n", "\n" ] }, { "data": { "text/plain": [ "somePrimaryKeys: scala.collection.immutable.Range = Range 1 to 10\n", "someMoreKeys: scala.collection.immutable.Range = Range 12 to 14\n", "batchGet2: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [__key: int, __digest: binary ... 7 more fields]\n" ] }, "execution_count": 11, "metadata": {}, "output_type": "execute_result" } ], "source": [ "//In this query we are doing *OR* between PK subqueries \n", "\n", "val somePrimaryKeys= 1.to(10).toSeq\n", "val someMoreKeys= 12.to(14).toSeq\n", "val batchGet2= spark2.sqlContext\n", ".read\n", ".format(\"aerospike\")\n", ".option(\"aerospike.set\", \"scala_input_data\")\n", ".option(\"aerospike.keyType\", \"int\") //used to hint primary key(PK) type when inferred without schema.\n", ".load.where((col(\"__key\") isin (somePrimaryKeys:_*)) || ( col(\"__key\") isin (someMoreKeys:_*) ))\n", "batchGet2.show(15)\n", "//We should got in total 13 records." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Queries that do not include Primary Key in the Predicate" ] }, { "cell_type": "code", "execution_count": 12, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----+--------------------+--------+------------+-----+---+-------+------+---+\n", "|__key| __digest|__expiry|__generation|__ttl|age| name|salary| id|\n", "+-----+--------------------+--------+------------+-----+---+-------+------+---+\n", "| 558|[14 80 A2 9D D2 E...| 0| 1| -1| 58|name558| 91583|558|\n", "| 687|[1A 30 21 88 39 A...| 0| 1| -1| 87|name687| 96887|687|\n", "| 372|[1B 40 51 DD 64 F...| 0| 1| -1| 72|name372| 64562|372|\n", "| 352|[23 A0 99 06 1F 7...| 0| 1| -1| 52|name352| 57325|352|\n", "| 94|[26 E0 C4 85 CE 9...| 0| 1| -1| 94| name94| 70982| 94|\n", "| 890|[26 30 F7 1A D3 A...| 0| 1| -1| 90|name890| 55053|890|\n", "| 163|[3E D0 72 42 15 9...| 0| 1| -1| 63|name163| 62598|163|\n", "| 882|[3E C0 28 CE F2 5...| 0| 1| -1| 82|name882| 63215|882|\n", "| 673|[45 10 C1 D6 80 3...| 0| 1| -1| 73|name673| 98636|673|\n", "| 991|[47 A0 D4 EC 12 1...| 0| 1| -1| 91|name991| 92557|991|\n", "| 293|[48 40 20 B0 E6 D...| 0| 1| -1| 93|name293| 82140|293|\n", "| 679|[57 80 24 4F 1D 3...| 0| 1| -1| 79|name679| 66054|679|\n", "| 153|[5D E0 05 75 BF 3...| 0| 1| -1| 53|name153| 79642|153|\n", "| 485|[6B 80 7E E1 A4 5...| 0| 1| -1| 85|name485| 54741|485|\n", "| 997|[72 10 81 9D E2 E...| 0| 1| -1| 97|name997| 80278|997|\n", "| 482|[85 B0 B1 3F 49 A...| 0| 1| -1| 82|name482| 79900|482|\n", "| 166|[8A 00 3E 64 19 D...| 0| 1| -1| 66|name166| 70542|166|\n", "| 590|[8C 20 A4 28 BE 7...| 0| 1| -1| 90|name590| 67077|590|\n", "| 689|[9B 00 70 22 F0 8...| 0| 1| -1| 89|name689| 93761|689|\n", "| 895|[9D A0 9D 91 AE 8...| 0| 1| -1| 95|name895| 76241|895|\n", "+-----+--------------------+--------+------------+-----+---+-------+------+---+\n", "only showing top 20 rows\n", "\n" ] }, { "data": { "text/plain": [ "somePrimaryKeys: scala.collection.immutable.Range = Range 1 to 10\n", "scanQuery1: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [__key: int, __digest: binary ... 7 more fields]\n" ] }, "execution_count": 12, "metadata": {}, "output_type": "execute_result" } ], "source": [ "\n", "val somePrimaryKeys= 1.to(10).toSeq\n", "val scanQuery1= spark2.sqlContext\n", ".read\n", ".format(\"aerospike\")\n", ".option(\"aerospike.set\", \"scala_input_data\")\n", ".option(\"aerospike.keyType\", \"int\") //used to hint primary key(PK) type when inferred without schema.\n", ".load.where((col(\"__key\") isin (somePrimaryKeys:_*)) || ( col(\"age\") >50 ))\n", "\n", "scanQuery1.show()\n", "\n", "//Since there is OR between PKs and Bin. It will be treated as Scan query. \n", "//Primary keys are not stored in bins(by default), hence only filters corresponding to bins are honored. " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Sampling from Aerospike DB\n", "\n", "- Sample specified number of records from Aerospike to considerably reduce data movement between Aerospike and the Spark clusters. Depending on the aerospike.partition.factor setting, you may get more records than desired. Please use this property in conjunction with Spark `limit()` function to get the specified number of records. The sample read is not randomized, so sample more than you need and use the Spark `sample()` function to randomize if you see fit. You can use it in conjunction with `aerospike.recordspersecond` to control the load on the Aerospike server while sampling.\n", "\n", "- For more information, please see [documentation](https://docs.aerospike.com/docs/connect/processing/spark/configuration.html) page.\n" ] }, { "cell_type": "code", "execution_count": 13, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "setname: String = scala_input_data\n", "sample_size: Int = 101\n", "df3: org.apache.spark.sql.DataFrame = [__key: string, __digest: binary ... 7 more fields]\n", "df4: org.apache.spark.sql.DataFrame = [__key: string, __digest: binary ... 7 more fields]\n", "count3: Long = 104\n", "count4: Long = 128\n", "dfWithLimit: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [__key: string, __digest: binary ... 7 more fields]\n", "limitCount: Long = 101\n" ] }, "execution_count": 13, "metadata": {}, "output_type": "execute_result" } ], "source": [ "//number_of_spark_partitions (num_sp)=2^{aerospike.partition.factor}\n", "//total number of records = Math.ceil((float)aerospike.sample.size/num_sp) * (num_sp) \n", "//use lower partition factor for more accurate sampling\n", "val setname=\"scala_input_data\"\n", "val sample_size=101\n", "\n", "val df3=spark2.read.format(\"aerospike\")\n", ".option(\"aerospike.partition.factor\",\"2\")\n", ".option(\"aerospike.set\",setname)\n", ".option(\"aerospike.sample.size\",\"101\") //allows to sample approximately spacific number of record. \n", ".load()\n", "\n", "val df4=spark2.read.format(\"aerospike\")\n", ".option(\"aerospike.partition.factor\",\"6\")\n", ".option(\"aerospike.set\",setname)\n", ".option(\"aerospike.sample.size\",\"101\") //allows to sample approximately spacific number of record. \n", ".load()\n", "\n", "//Notice that more records were read than requested due to the underlying partitioning logic related to the partition factor as described earlier, hence we use Spark limit() function additionally to return the desired number of records.\n", "val count3=df3.count()\n", "val count4=df4.count()\n", "\n", "\n", "//Note how limit got only 101 record from df4 which have 128 records.\n", "val dfWithLimit=df4.limit(101)\n", "val limitCount=dfWithLimit.count()" ] } ], "metadata": { "kernelspec": { "display_name": "spylon-kernel", "language": "scala", "name": "spylon-kernel" }, "language_info": { "codemirror_mode": "text/x-scala", "file_extension": ".scala", "help_links": [ { "text": "MetaKernel Magics", "url": "https://metakernel.readthedocs.io/en/latest/source/README.html" } ], "mimetype": "text/x-scala", "name": "scala", "pygments_lexer": "scala", "version": "0.4.1" } }, "nbformat": 4, "nbformat_minor": 2 }