notebooks/01-spark-struct-stream-kafka.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "# Spark Structured Streaming from Kafka\n",
    "\n",
    "Example of using Spark to connect to Kafka and using Spark Structured Streaming to process a Kafka stream of Python alerts in non-Avro string format."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "## Notes\n",
    "\n",
    "Useful references.\n",
    "\n",
    "https://spark.apache.org/docs/2.1.0/structured-streaming-kafka-integration.html\n",
    "\n",
    "https://spark.apache.org/docs/2.1.0/structured-streaming-programming-guide.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "source": [
    "## Prep environment\n",
    "\n",
    "Need some packages to talk to Kafka."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.1.0,org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0 pyspark-shell'\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "from ast import literal_eval"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "## Start a Kafka stream for Spark to subscribe"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "With lsst-dm/alert_stream, in an external shell:\n",
    "\n",
    "Send some alerts so stream exists to connect to:\n",
    "\n",
    "docker run -it       --network=alertstream_default       alert_stream python bin/sendAlertStream.py my-stream 10 --no-stamps --encode-off"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "## Set up Spark session and connection to Kakfa"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"SSKafka\") \\\n",
    "    .getOrCreate()\n",
    "    \n",
    "# default for startingOffsets is \"latest\", but \"earliest\" allows rewind for missed alerts    \n",
    "dsraw = spark \\\n",
    "  .readStream \\\n",
    "  .format(\"kafka\") \\\n",
    "  .option(\"kafka.bootstrap.servers\", \"kafka:9092\") \\\n",
    "  .option(\"subscribe\", \"my-stream\") \\\n",
    "  .option(\"startingOffsets\", \"earliest\") \\\n",
    "  .load()\n",
    "  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "ds = dsraw.selectExpr(\"CAST(value AS STRING)\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pyspark.sql.dataframe.DataFrame'>\n",
      "<class 'pyspark.sql.dataframe.DataFrame'>\n"
     ]
    }
   ],
   "source": [
    "print(type(dsraw))\n",
    "print(type(ds))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "dsraw is the raw data stream, in \"kafka\" format.\n",
    "\n",
    "ds pulls out the \"value\" from \"kafka\" format, the actual alert data."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "## Create output for Spark Structured Streaming\n",
    "\n",
    "Queries are new sql dataframe streams and can be written to disk or saved to memory for followup sql operations.\n",
    "Below they are saved to memory with queryNames that can be treated as tables by spark.sql.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "rawQuery = dsraw \\\n",
    "        .writeStream \\\n",
    "        .queryName(\"qraw\")\\\n",
    "        .format(\"memory\")\\\n",
    "        .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "alertQuery = ds \\\n",
    "        .writeStream \\\n",
    "        .queryName(\"qalerts\")\\\n",
    "        .format(\"memory\")\\\n",
    "        .start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "Send some alerts to get queries to recognize activity:\n",
    "\n",
    "docker run -it       --network=alertstream_default       alert_stream python bin/sendAlertStream.py my-stream 10 --no-stamps --encode-off"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use sql operations on the named in-memory query tables."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------------+---------+---------+------+--------------------+-------------+\n",
      "| key|               value|    topic|partition|offset|           timestamp|timestampType|\n",
      "+----+--------------------+---------+---------+------+--------------------+-------------+\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|     0|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|     1|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|     2|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|     3|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|     4|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|     5|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|     6|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|     7|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|     8|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|     9|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|    10|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|    11|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|    12|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|    13|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|    14|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|    15|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|    16|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|    17|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|    18|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 27 61 6C 65 7...|my-stream|        0|    19|1969-12-31 23:59:...|            0|\n",
      "+----+--------------------+---------+---------+------+--------------------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "raw = spark.sql(\"select * from qraw\")\n",
    "raw.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Because this stream is format=\"kafka,\" the schema of the table reflects the data structure of Kafka streams, not of our data content, which is stored in \"value.\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: binary (nullable = true)\n",
      " |-- value: binary (nullable = true)\n",
      " |-- topic: string (nullable = true)\n",
      " |-- partition: integer (nullable = true)\n",
      " |-- offset: long (nullable = true)\n",
      " |-- timestamp: timestamp (nullable = true)\n",
      " |-- timestampType: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "raw.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Try selecting from the stream that has cast the kafka \"value\" to strings."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|               value|\n",
      "+--------------------+\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "|{'alertId': 12313...|\n",
      "+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "alerts = spark.sql(\"select * from qalerts\")\n",
    "alerts.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The alert data has no known schema, only str."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- value: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "alerts.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(alerts)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Convert the values in the sql dataframe to RDD of dicts\n",
    "\n",
    "We can use the RDDs to convert data to a better structure for filtering."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "rddAlertsRdd = alerts.rdd.map(lambda alert: literal_eval(alert['value']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[24] at RDD at PythonRDD.scala:48"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rddAlertsRdd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rddAlerts = rddAlertsRdd.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "list"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(rddAlerts)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'alertId': 1231321321,\n",
       " 'diaObject': {'decl': 0.126243049656,\n",
       "  'diaObjectId': 281323062375219201,\n",
       "  'flags': 0,\n",
       "  'parallax': 2.124124,\n",
       "  'pmDecl': 0.00014,\n",
       "  'pmParallaxChi2': 0.00013,\n",
       "  'pmParallaxLnL': 0.00013,\n",
       "  'pmParallaxNdata': 1214,\n",
       "  'pmRa': 0.00013,\n",
       "  'pm_parallax_Cov': {'parallaxSigma': 0.00013,\n",
       "   'pmDeclSigma': 0.00013,\n",
       "   'pmDecl_parallax_Cov': 0.00013,\n",
       "   'pmRaSigma': 0.00013,\n",
       "   'pmRa_parallax_Cov': 0.00013,\n",
       "   'pmRa_pmDecl_Cov': 0.00013},\n",
       "  'ra': 351.570546978,\n",
       "  'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "   'raSigma': 0.00028,\n",
       "   'ra_decl_Cov': 0.00029},\n",
       "  'radecTai': 1480360995},\n",
       " 'diaObjectL2': {'decl': 0.126243049656,\n",
       "  'diaObjectId': 281323062375219201,\n",
       "  'flags': 0,\n",
       "  'parallax': 2.124124,\n",
       "  'pmDecl': 0.00014,\n",
       "  'pmParallaxChi2': 0.00013,\n",
       "  'pmParallaxLnL': 0.00013,\n",
       "  'pmParallaxNdata': 1214,\n",
       "  'pmRa': 0.00013,\n",
       "  'pm_parallax_Cov': {'parallaxSigma': 0.00013,\n",
       "   'pmDeclSigma': 0.00013,\n",
       "   'pmDecl_parallax_Cov': 0.00013,\n",
       "   'pmRaSigma': 0.00013,\n",
       "   'pmRa_parallax_Cov': 0.00013,\n",
       "   'pmRa_pmDecl_Cov': 0.00013},\n",
       "  'ra': 351.570546978,\n",
       "  'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "   'raSigma': 0.00028,\n",
       "   'ra_decl_Cov': 0.00029},\n",
       "  'radecTai': 1480360995},\n",
       " 'diaSource': {'ccdVisitId': 111111,\n",
       "  'decl': 0.126243049656,\n",
       "  'diaSourceId': 281323062375219200,\n",
       "  'filterName': 'r',\n",
       "  'flags': 0,\n",
       "  'midPointTai': 1480360995,\n",
       "  'psFlux': 1241.0,\n",
       "  'ra': 351.570546978,\n",
       "  'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "   'raSigma': 0.00028,\n",
       "   'ra_decl_Cov': 0.00029},\n",
       "  'snr': 41.1,\n",
       "  'x': 112.1,\n",
       "  'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "  'y': 121.1},\n",
       " 'diaSourcesL2': [{'ccdVisitId': 111111,\n",
       "   'decl': 0.126243049656,\n",
       "   'diaSourceId': 281323062375219198,\n",
       "   'filterName': 'r',\n",
       "   'flags': 0,\n",
       "   'midPointTai': 1480360995,\n",
       "   'psFlux': 1241.0,\n",
       "   'ra': 351.570546978,\n",
       "   'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "    'raSigma': 0.00028,\n",
       "    'ra_decl_Cov': 0.00029},\n",
       "   'snr': 41.1,\n",
       "   'x': 112.1,\n",
       "   'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "   'y': 121.1},\n",
       "  {'ccdVisitId': 111111,\n",
       "   'decl': 0.126243049656,\n",
       "   'diaSourceId': 281323062375219199,\n",
       "   'filterName': 'r',\n",
       "   'flags': 0,\n",
       "   'midPointTai': 1480360995,\n",
       "   'psFlux': 1241.0,\n",
       "   'ra': 351.570546978,\n",
       "   'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "    'raSigma': 0.00028,\n",
       "    'ra_decl_Cov': 0.00029},\n",
       "   'snr': 41.1,\n",
       "   'x': 112.1,\n",
       "   'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "   'y': 121.1}],\n",
       " 'l1dbId': 222222222,\n",
       " 'prv_diaSources': [{'ccdVisitId': 111111,\n",
       "   'decl': 0.126243049656,\n",
       "   'diaSourceId': 281323062375219198,\n",
       "   'filterName': 'r',\n",
       "   'flags': 0,\n",
       "   'midPointTai': 1480360995,\n",
       "   'psFlux': 1241.0,\n",
       "   'ra': 351.570546978,\n",
       "   'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "    'raSigma': 0.00028,\n",
       "    'ra_decl_Cov': 0.00029},\n",
       "   'snr': 41.1,\n",
       "   'x': 112.1,\n",
       "   'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "   'y': 121.1},\n",
       "  {'ccdVisitId': 111111,\n",
       "   'decl': 0.126243049656,\n",
       "   'diaSourceId': 281323062375219199,\n",
       "   'filterName': 'r',\n",
       "   'flags': 0,\n",
       "   'midPointTai': 1480360995,\n",
       "   'psFlux': 1241.0,\n",
       "   'ra': 351.570546978,\n",
       "   'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "    'raSigma': 0.00028,\n",
       "    'ra_decl_Cov': 0.00029},\n",
       "   'snr': 41.1,\n",
       "   'x': 112.1,\n",
       "   'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "   'y': 121.1}],\n",
       " 'ssObject': {'MOID1': 3141.0,\n",
       "  'MOID2': 23432.423,\n",
       "  'arc': 2.124124,\n",
       "  'flags': 0,\n",
       "  'moidLon1': 2143.213,\n",
       "  'moidLon2': 3142.23123,\n",
       "  'oe': {'M': 131.1241,\n",
       "   'aop': 344243.3,\n",
       "   'e': 636.121,\n",
       "   'epoch': 134141,\n",
       "   'i': 5436.2344,\n",
       "   'lan': 54325.34,\n",
       "   'q': 6654.14},\n",
       "  'orbFitChi2': 1341421.2414,\n",
       "  'orbFitLnL': 1343141.0341,\n",
       "  'orbFitNdata': 1214,\n",
       "  'ssObjectId': 5364546,\n",
       "  'uG1': 32131.312,\n",
       "  'uG1Err': 31232.2132,\n",
       "  'uG2': 231.2313,\n",
       "  'uG2Err': 23132.231,\n",
       "  'uH': 13231.231,\n",
       "  'uHErr': 13213.213}}"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rddAlerts[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Do filtering as list comprehension."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "ra_all = [alert for alert in rddAlerts \\\n",
    "                 if alert['diaSource']['ra'] > 350 ]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "20\n"
     ]
    }
   ],
   "source": [
    "print(len(ra_all))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'alertId': 1231321321,\n",
       "  'diaObject': {'decl': 0.126243049656,\n",
       "   'diaObjectId': 281323062375219201,\n",
       "   'flags': 0,\n",
       "   'parallax': 2.124124,\n",
       "   'pmDecl': 0.00014,\n",
       "   'pmParallaxChi2': 0.00013,\n",
       "   'pmParallaxLnL': 0.00013,\n",
       "   'pmParallaxNdata': 1214,\n",
       "   'pmRa': 0.00013,\n",
       "   'pm_parallax_Cov': {'parallaxSigma': 0.00013,\n",
       "    'pmDeclSigma': 0.00013,\n",
       "    'pmDecl_parallax_Cov': 0.00013,\n",
       "    'pmRaSigma': 0.00013,\n",
       "    'pmRa_parallax_Cov': 0.00013,\n",
       "    'pmRa_pmDecl_Cov': 0.00013},\n",
       "   'ra': 351.570546978,\n",
       "   'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "    'raSigma': 0.00028,\n",
       "    'ra_decl_Cov': 0.00029},\n",
       "   'radecTai': 1480360995},\n",
       "  'diaObjectL2': {'decl': 0.126243049656,\n",
       "   'diaObjectId': 281323062375219201,\n",
       "   'flags': 0,\n",
       "   'parallax': 2.124124,\n",
       "   'pmDecl': 0.00014,\n",
       "   'pmParallaxChi2': 0.00013,\n",
       "   'pmParallaxLnL': 0.00013,\n",
       "   'pmParallaxNdata': 1214,\n",
       "   'pmRa': 0.00013,\n",
       "   'pm_parallax_Cov': {'parallaxSigma': 0.00013,\n",
       "    'pmDeclSigma': 0.00013,\n",
       "    'pmDecl_parallax_Cov': 0.00013,\n",
       "    'pmRaSigma': 0.00013,\n",
       "    'pmRa_parallax_Cov': 0.00013,\n",
       "    'pmRa_pmDecl_Cov': 0.00013},\n",
       "   'ra': 351.570546978,\n",
       "   'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "    'raSigma': 0.00028,\n",
       "    'ra_decl_Cov': 0.00029},\n",
       "   'radecTai': 1480360995},\n",
       "  'diaSource': {'ccdVisitId': 111111,\n",
       "   'decl': 0.126243049656,\n",
       "   'diaSourceId': 281323062375219200,\n",
       "   'filterName': 'r',\n",
       "   'flags': 0,\n",
       "   'midPointTai': 1480360995,\n",
       "   'psFlux': 1241.0,\n",
       "   'ra': 351.570546978,\n",
       "   'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "    'raSigma': 0.00028,\n",
       "    'ra_decl_Cov': 0.00029},\n",
       "   'snr': 41.1,\n",
       "   'x': 112.1,\n",
       "   'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "   'y': 121.1},\n",
       "  'diaSourcesL2': [{'ccdVisitId': 111111,\n",
       "    'decl': 0.126243049656,\n",
       "    'diaSourceId': 281323062375219198,\n",
       "    'filterName': 'r',\n",
       "    'flags': 0,\n",
       "    'midPointTai': 1480360995,\n",
       "    'psFlux': 1241.0,\n",
       "    'ra': 351.570546978,\n",
       "    'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "     'raSigma': 0.00028,\n",
       "     'ra_decl_Cov': 0.00029},\n",
       "    'snr': 41.1,\n",
       "    'x': 112.1,\n",
       "    'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "    'y': 121.1},\n",
       "   {'ccdVisitId': 111111,\n",
       "    'decl': 0.126243049656,\n",
       "    'diaSourceId': 281323062375219199,\n",
       "    'filterName': 'r',\n",
       "    'flags': 0,\n",
       "    'midPointTai': 1480360995,\n",
       "    'psFlux': 1241.0,\n",
       "    'ra': 351.570546978,\n",
       "    'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "     'raSigma': 0.00028,\n",
       "     'ra_decl_Cov': 0.00029},\n",
       "    'snr': 41.1,\n",
       "    'x': 112.1,\n",
       "    'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "    'y': 121.1}],\n",
       "  'l1dbId': 222222222,\n",
       "  'prv_diaSources': [{'ccdVisitId': 111111,\n",
       "    'decl': 0.126243049656,\n",
       "    'diaSourceId': 281323062375219198,\n",
       "    'filterName': 'r',\n",
       "    'flags': 0,\n",
       "    'midPointTai': 1480360995,\n",
       "    'psFlux': 1241.0,\n",
       "    'ra': 351.570546978,\n",
       "    'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "     'raSigma': 0.00028,\n",
       "     'ra_decl_Cov': 0.00029},\n",
       "    'snr': 41.1,\n",
       "    'x': 112.1,\n",
       "    'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "    'y': 121.1},\n",
       "   {'ccdVisitId': 111111,\n",
       "    'decl': 0.126243049656,\n",
       "    'diaSourceId': 281323062375219199,\n",
       "    'filterName': 'r',\n",
       "    'flags': 0,\n",
       "    'midPointTai': 1480360995,\n",
       "    'psFlux': 1241.0,\n",
       "    'ra': 351.570546978,\n",
       "    'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "     'raSigma': 0.00028,\n",
       "     'ra_decl_Cov': 0.00029},\n",
       "    'snr': 41.1,\n",
       "    'x': 112.1,\n",
       "    'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "    'y': 121.1}],\n",
       "  'ssObject': {'MOID1': 3141.0,\n",
       "   'MOID2': 23432.423,\n",
       "   'arc': 2.124124,\n",
       "   'flags': 0,\n",
       "   'moidLon1': 2143.213,\n",
       "   'moidLon2': 3142.23123,\n",
       "   'oe': {'M': 131.1241,\n",
       "    'aop': 344243.3,\n",
       "    'e': 636.121,\n",
       "    'epoch': 134141,\n",
       "    'i': 5436.2344,\n",
       "    'lan': 54325.34,\n",
       "    'q': 6654.14},\n",
       "   'orbFitChi2': 1341421.2414,\n",
       "   'orbFitLnL': 1343141.0341,\n",
       "   'orbFitNdata': 1214,\n",
       "   'ssObjectId': 5364546,\n",
       "   'uG1': 32131.312,\n",
       "   'uG1Err': 31232.2132,\n",
       "   'uG2': 231.2313,\n",
       "   'uG2Err': 23132.231,\n",
       "   'uH': 13231.231,\n",
       "   'uHErr': 13213.213}},\n",
       " {'alertId': 1231321321,\n",
       "  'diaObject': {'decl': 0.126243049656,\n",
       "   'diaObjectId': 281323062375219201,\n",
       "   'flags': 0,\n",
       "   'parallax': 2.124124,\n",
       "   'pmDecl': 0.00014,\n",
       "   'pmParallaxChi2': 0.00013,\n",
       "   'pmParallaxLnL': 0.00013,\n",
       "   'pmParallaxNdata': 1214,\n",
       "   'pmRa': 0.00013,\n",
       "   'pm_parallax_Cov': {'parallaxSigma': 0.00013,\n",
       "    'pmDeclSigma': 0.00013,\n",
       "    'pmDecl_parallax_Cov': 0.00013,\n",
       "    'pmRaSigma': 0.00013,\n",
       "    'pmRa_parallax_Cov': 0.00013,\n",
       "    'pmRa_pmDecl_Cov': 0.00013},\n",
       "   'ra': 351.570546978,\n",
       "   'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "    'raSigma': 0.00028,\n",
       "    'ra_decl_Cov': 0.00029},\n",
       "   'radecTai': 1480360995},\n",
       "  'diaObjectL2': {'decl': 0.126243049656,\n",
       "   'diaObjectId': 281323062375219201,\n",
       "   'flags': 0,\n",
       "   'parallax': 2.124124,\n",
       "   'pmDecl': 0.00014,\n",
       "   'pmParallaxChi2': 0.00013,\n",
       "   'pmParallaxLnL': 0.00013,\n",
       "   'pmParallaxNdata': 1214,\n",
       "   'pmRa': 0.00013,\n",
       "   'pm_parallax_Cov': {'parallaxSigma': 0.00013,\n",
       "    'pmDeclSigma': 0.00013,\n",
       "    'pmDecl_parallax_Cov': 0.00013,\n",
       "    'pmRaSigma': 0.00013,\n",
       "    'pmRa_parallax_Cov': 0.00013,\n",
       "    'pmRa_pmDecl_Cov': 0.00013},\n",
       "   'ra': 351.570546978,\n",
       "   'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "    'raSigma': 0.00028,\n",
       "    'ra_decl_Cov': 0.00029},\n",
       "   'radecTai': 1480360995},\n",
       "  'diaSource': {'ccdVisitId': 111111,\n",
       "   'decl': 0.126243049656,\n",
       "   'diaSourceId': 281323062375219200,\n",
       "   'filterName': 'r',\n",
       "   'flags': 0,\n",
       "   'midPointTai': 1480360995,\n",
       "   'psFlux': 1241.0,\n",
       "   'ra': 351.570546978,\n",
       "   'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "    'raSigma': 0.00028,\n",
       "    'ra_decl_Cov': 0.00029},\n",
       "   'snr': 41.1,\n",
       "   'x': 112.1,\n",
       "   'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "   'y': 121.1},\n",
       "  'diaSourcesL2': [{'ccdVisitId': 111111,\n",
       "    'decl': 0.126243049656,\n",
       "    'diaSourceId': 281323062375219198,\n",
       "    'filterName': 'r',\n",
       "    'flags': 0,\n",
       "    'midPointTai': 1480360995,\n",
       "    'psFlux': 1241.0,\n",
       "    'ra': 351.570546978,\n",
       "    'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "     'raSigma': 0.00028,\n",
       "     'ra_decl_Cov': 0.00029},\n",
       "    'snr': 41.1,\n",
       "    'x': 112.1,\n",
       "    'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "    'y': 121.1},\n",
       "   {'ccdVisitId': 111111,\n",
       "    'decl': 0.126243049656,\n",
       "    'diaSourceId': 281323062375219199,\n",
       "    'filterName': 'r',\n",
       "    'flags': 0,\n",
       "    'midPointTai': 1480360995,\n",
       "    'psFlux': 1241.0,\n",
       "    'ra': 351.570546978,\n",
       "    'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "     'raSigma': 0.00028,\n",
       "     'ra_decl_Cov': 0.00029},\n",
       "    'snr': 41.1,\n",
       "    'x': 112.1,\n",
       "    'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "    'y': 121.1}],\n",
       "  'l1dbId': 222222222,\n",
       "  'prv_diaSources': [{'ccdVisitId': 111111,\n",
       "    'decl': 0.126243049656,\n",
       "    'diaSourceId': 281323062375219198,\n",
       "    'filterName': 'r',\n",
       "    'flags': 0,\n",
       "    'midPointTai': 1480360995,\n",
       "    'psFlux': 1241.0,\n",
       "    'ra': 351.570546978,\n",
       "    'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "     'raSigma': 0.00028,\n",
       "     'ra_decl_Cov': 0.00029},\n",
       "    'snr': 41.1,\n",
       "    'x': 112.1,\n",
       "    'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "    'y': 121.1},\n",
       "   {'ccdVisitId': 111111,\n",
       "    'decl': 0.126243049656,\n",
       "    'diaSourceId': 281323062375219199,\n",
       "    'filterName': 'r',\n",
       "    'flags': 0,\n",
       "    'midPointTai': 1480360995,\n",
       "    'psFlux': 1241.0,\n",
       "    'ra': 351.570546978,\n",
       "    'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "     'raSigma': 0.00028,\n",
       "     'ra_decl_Cov': 0.00029},\n",
       "    'snr': 41.1,\n",
       "    'x': 112.1,\n",
       "    'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       "    'y': 121.1}],\n",
       "  'ssObject': {'MOID1': 3141.0,\n",
       "   'MOID2': 23432.423,\n",
       "   'arc': 2.124124,\n",
       "   'flags': 0,\n",
       "   'moidLon1': 2143.213,\n",
       "   'moidLon2': 3142.23123,\n",
       "   'oe': {'M': 131.1241,\n",
       "    'aop': 344243.3,\n",
       "    'e': 636.121,\n",
       "    'epoch': 134141,\n",
       "    'i': 5436.2344,\n",
       "    'lan': 54325.34,\n",
       "    'q': 6654.14},\n",
       "   'orbFitChi2': 1341421.2414,\n",
       "   'orbFitLnL': 1343141.0341,\n",
       "   'orbFitNdata': 1214,\n",
       "   'ssObjectId': 5364546,\n",
       "   'uG1': 32131.312,\n",
       "   'uG1Err': 31232.2132,\n",
       "   'uG2': 231.2313,\n",
       "   'uG2Err': 23132.231,\n",
       "   'uH': 13231.231,\n",
       "   'uHErr': 13213.213}}]"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ra_all[0:2]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "ra_empty = [alert for alert in rddAlerts \\\n",
    "                 if alert['diaSource']['ra'] < 350 ]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0\n"
     ]
    }
   ],
   "source": [
    "print(len(ra_empty))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Converting from RDD of dicts to a dataframe is not a good idea\n",
    "\n",
    "The schema is inferred incorrectly, and data can be lost, shown below.  So, don't try to do filtering with sql dataframes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/spark/python/pyspark/sql/session.py:336: UserWarning: Using RDD of dict to inferSchema is deprecated. Use pyspark.sql.Row instead\n",
      "  warnings.warn(\"Using RDD of dict to inferSchema is deprecated. \"\n"
     ]
    }
   ],
   "source": [
    "df = rddAlertsRdd.toDF()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+--------------------+--------------------+--------------------+---------+--------------------+--------------------+\n",
      "|   alertId|           diaObject|         diaObjectL2|           diaSource|        diaSourcesL2|   l1dbId|      prv_diaSources|            ssObject|\n",
      "+----------+--------------------+--------------------+--------------------+--------------------+---------+--------------------+--------------------+\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "|1231321321|Map(decl -> 0.126...|Map(decl -> 0.126...|Map(x -> null, cc...|[Map(x -> null, c...|222222222|[Map(x -> null, c...|Map(orbFitNdata -...|\n",
      "+----------+--------------------+--------------------+--------------------+--------------------+---------+--------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "alertIds = df.select('alertId')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|   alertId|\n",
      "+----------+\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "|1231321321|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "alertIds.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Filtering _appears_ to be working above, for the data that is not lost.  Below shows NULLs where data has been lost."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "diaSources = df.select('diaSource').where('diaSource.diaSourceId is not NULL')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "diaSources_empty = df.select('diaSource').where('diaSource.diaSourceId is NULL')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(diaSources)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+\n",
      "|diaSource|\n",
      "+---------+\n",
      "+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "diaSources.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|           diaSource|\n",
      "+--------------------+\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "diaSources_empty.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Take a closer look at diaSources_empty with a pandas dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "pd = diaSources_empty.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>diaSource</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>10</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>11</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>12</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>13</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>14</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>15</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>16</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>17</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>18</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>19</th>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                            diaSource\n",
       "0   {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "1   {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "2   {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "3   {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "4   {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "5   {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "6   {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "7   {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "8   {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "9   {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "10  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "11  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "12  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "13  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "14  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "15  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "16  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "17  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "18  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...\n",
       "19  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_..."
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dict"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(pd['diaSource'][0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'ccdVisitId': None,\n",
       " 'decl': None,\n",
       " 'diaSourceId': None,\n",
       " 'filterName': None,\n",
       " 'flags': None,\n",
       " 'midPointTai': None,\n",
       " 'psFlux': None,\n",
       " 'ra': None,\n",
       " 'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "  'raSigma': 0.00028,\n",
       "  'ra_decl_Cov': 0.00029},\n",
       " 'snr': None,\n",
       " 'x': None,\n",
       " 'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       " 'y': None}"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd['diaSource'][0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Some data has been misinterpreted, shown by the \"None\"s above."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "What if we try from the pre-pandas sql dataframe?  Checking to see if it was the pandas conversion that lost data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|           diaSource|\n",
      "+--------------------+\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "|Map(x -> null, cc...|\n",
      "+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "diaSources_empty.filter('diaSource.filterName is NULL').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "No, shown above, data was lost before the pandas conversion.  Don't do RDD.toDF() when RDD is dicts."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Original sql.dataframe to pandas is ok"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "pdAlerts = alerts.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\"{'alertId': 1231321321, 'l1dbId': 222222222, 'diaSource': {'diaSourceId': 281323062375219200, 'ccdVisitId': 111111, 'midPointTai': 1480360995, 'filterName': 'r', 'ra': 351.570546978, 'decl': 0.126243049656, 'ra_decl_Cov': {'raSigma': 0.00028, 'declSigma': 0.00028, 'ra_decl_Cov': 0.00029}, 'x': 112.1, 'y': 121.1, 'x_y_Cov': {'xSigma': 1.2, 'ySigma': 1.1, 'x_y_Cov': 1.2}, 'snr': 41.1, 'psFlux': 1241.0, 'flags': 0}, 'prv_diaSources': [{'diaSourceId': 281323062375219198, 'ccdVisitId': 111111, 'midPointTai': 1480360995, 'filterName': 'r', 'ra': 351.570546978, 'decl': 0.126243049656, 'ra_decl_Cov': {'raSigma': 0.00028, 'declSigma': 0.00028, 'ra_decl_Cov': 0.00029}, 'x': 112.1, 'y': 121.1, 'x_y_Cov': {'xSigma': 1.2, 'ySigma': 1.1, 'x_y_Cov': 1.2}, 'snr': 41.1, 'psFlux': 1241.0, 'flags': 0}, {'diaSourceId': 281323062375219199, 'ccdVisitId': 111111, 'midPointTai': 1480360995, 'filterName': 'r', 'ra': 351.570546978, 'decl': 0.126243049656, 'ra_decl_Cov': {'raSigma': 0.00028, 'declSigma': 0.00028, 'ra_decl_Cov': 0.00029}, 'x': 112.1, 'y': 121.1, 'x_y_Cov': {'xSigma': 1.2, 'ySigma': 1.1, 'x_y_Cov': 1.2}, 'snr': 41.1, 'psFlux': 1241.0, 'flags': 0}], 'diaObject': {'diaObjectId': 281323062375219201, 'ra': 351.570546978, 'decl': 0.126243049656, 'ra_decl_Cov': {'raSigma': 0.00028, 'declSigma': 0.00028, 'ra_decl_Cov': 0.00029}, 'radecTai': 1480360995, 'pmRa': 0.00013, 'pmDecl': 0.00014, 'parallax': 2.124124, 'pm_parallax_Cov': {'pmRaSigma': 0.00013, 'pmDeclSigma': 0.00013, 'parallaxSigma': 0.00013, 'pmRa_pmDecl_Cov': 0.00013, 'pmRa_parallax_Cov': 0.00013, 'pmDecl_parallax_Cov': 0.00013}, 'pmParallaxLnL': 0.00013, 'pmParallaxChi2': 0.00013, 'pmParallaxNdata': 1214, 'flags': 0}, 'ssObject': {'ssObjectId': 5364546, 'oe': {'q': 6654.14, 'e': 636.121, 'i': 5436.2344, 'lan': 54325.34, 'aop': 344243.3, 'M': 131.1241, 'epoch': 134141}, 'arc': 2.124124, 'orbFitLnL': 1343141.0341, 'orbFitChi2': 1341421.2414, 'orbFitNdata': 1214, 'MOID1': 3141.0, 'MOID2': 23432.423, 'moidLon1': 2143.213, 'moidLon2': 3142.23123, 'uH': 13231.231, 'uHErr': 13213.213, 'uG1': 32131.312, 'uG1Err': 31232.2132, 'uG2': 231.2313, 'uG2Err': 23132.231, 'flags': 0}, 'diaObjectL2': {'diaObjectId': 281323062375219201, 'ra': 351.570546978, 'decl': 0.126243049656, 'ra_decl_Cov': {'raSigma': 0.00028, 'declSigma': 0.00028, 'ra_decl_Cov': 0.00029}, 'radecTai': 1480360995, 'pmRa': 0.00013, 'pmDecl': 0.00014, 'parallax': 2.124124, 'pm_parallax_Cov': {'pmRaSigma': 0.00013, 'pmDeclSigma': 0.00013, 'parallaxSigma': 0.00013, 'pmRa_pmDecl_Cov': 0.00013, 'pmRa_parallax_Cov': 0.00013, 'pmDecl_parallax_Cov': 0.00013}, 'pmParallaxLnL': 0.00013, 'pmParallaxChi2': 0.00013, 'pmParallaxNdata': 1214, 'flags': 0}, 'diaSourcesL2': [{'diaSourceId': 281323062375219198, 'ccdVisitId': 111111, 'midPointTai': 1480360995, 'filterName': 'r', 'ra': 351.570546978, 'decl': 0.126243049656, 'ra_decl_Cov': {'raSigma': 0.00028, 'declSigma': 0.00028, 'ra_decl_Cov': 0.00029}, 'x': 112.1, 'y': 121.1, 'x_y_Cov': {'xSigma': 1.2, 'ySigma': 1.1, 'x_y_Cov': 1.2}, 'snr': 41.1, 'psFlux': 1241.0, 'flags': 0}, {'diaSourceId': 281323062375219199, 'ccdVisitId': 111111, 'midPointTai': 1480360995, 'filterName': 'r', 'ra': 351.570546978, 'decl': 0.126243049656, 'ra_decl_Cov': {'raSigma': 0.00028, 'declSigma': 0.00028, 'ra_decl_Cov': 0.00029}, 'x': 112.1, 'y': 121.1, 'x_y_Cov': {'xSigma': 1.2, 'ySigma': 1.1, 'x_y_Cov': 1.2}, 'snr': 41.1, 'psFlux': 1241.0, 'flags': 0}]}\""
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pdAlerts['value'][0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "str"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(pdAlerts['value'][0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "pdAlertsSeries = pdAlerts['value'].map(literal_eval)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pandas.core.series.Series"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(pdAlertsSeries)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0    {'prv_diaSources': [{'ra_decl_Cov': {'raSigma'...\n",
       "1    {'prv_diaSources': [{'ra_decl_Cov': {'raSigma'...\n",
       "2    {'prv_diaSources': [{'ra_decl_Cov': {'raSigma'...\n",
       "Name: value, dtype: object"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pdAlertsSeries[0:3]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dict"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(pdAlertsSeries[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "import pandas\n",
    "pdAlertsDf = pandas.DataFrame(list(pdAlertsSeries))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pandas.core.frame.DataFrame"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(pdAlertsDf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>alertId</th>\n",
       "      <th>diaObject</th>\n",
       "      <th>diaObjectL2</th>\n",
       "      <th>diaSource</th>\n",
       "      <th>diaSourcesL2</th>\n",
       "      <th>l1dbId</th>\n",
       "      <th>prv_diaSources</th>\n",
       "      <th>ssObject</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1231321321</td>\n",
       "      <td>{'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...</td>\n",
       "      <td>{'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...</td>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "      <td>[{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...</td>\n",
       "      <td>222222222</td>\n",
       "      <td>[{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...</td>\n",
       "      <td>{'uG2': 231.2313, 'arc': 2.124124, 'uG2Err': 2...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1231321321</td>\n",
       "      <td>{'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...</td>\n",
       "      <td>{'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...</td>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "      <td>[{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...</td>\n",
       "      <td>222222222</td>\n",
       "      <td>[{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...</td>\n",
       "      <td>{'uG2': 231.2313, 'arc': 2.124124, 'uG2Err': 2...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1231321321</td>\n",
       "      <td>{'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...</td>\n",
       "      <td>{'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...</td>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "      <td>[{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...</td>\n",
       "      <td>222222222</td>\n",
       "      <td>[{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...</td>\n",
       "      <td>{'uG2': 231.2313, 'arc': 2.124124, 'uG2Err': 2...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1231321321</td>\n",
       "      <td>{'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...</td>\n",
       "      <td>{'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...</td>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "      <td>[{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...</td>\n",
       "      <td>222222222</td>\n",
       "      <td>[{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...</td>\n",
       "      <td>{'uG2': 231.2313, 'arc': 2.124124, 'uG2Err': 2...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1231321321</td>\n",
       "      <td>{'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...</td>\n",
       "      <td>{'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...</td>\n",
       "      <td>{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...</td>\n",
       "      <td>[{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...</td>\n",
       "      <td>222222222</td>\n",
       "      <td>[{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...</td>\n",
       "      <td>{'uG2': 231.2313, 'arc': 2.124124, 'uG2Err': 2...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "      alertId                                          diaObject  \\\n",
       "0  1231321321  {'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...   \n",
       "1  1231321321  {'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...   \n",
       "2  1231321321  {'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...   \n",
       "3  1231321321  {'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...   \n",
       "4  1231321321  {'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...   \n",
       "\n",
       "                                         diaObjectL2  \\\n",
       "0  {'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...   \n",
       "1  {'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...   \n",
       "2  {'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...   \n",
       "3  {'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...   \n",
       "4  {'pmRa': 0.00013, 'pmParallaxNdata': 1214, 'pm...   \n",
       "\n",
       "                                           diaSource  \\\n",
       "0  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...   \n",
       "1  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...   \n",
       "2  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...   \n",
       "3  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...   \n",
       "4  {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_...   \n",
       "\n",
       "                                        diaSourcesL2     l1dbId  \\\n",
       "0  [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...  222222222   \n",
       "1  [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...  222222222   \n",
       "2  [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...  222222222   \n",
       "3  [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...  222222222   \n",
       "4  [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...  222222222   \n",
       "\n",
       "                                      prv_diaSources  \\\n",
       "0  [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...   \n",
       "1  [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...   \n",
       "2  [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...   \n",
       "3  [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...   \n",
       "4  [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl...   \n",
       "\n",
       "                                            ssObject  \n",
       "0  {'uG2': 231.2313, 'arc': 2.124124, 'uG2Err': 2...  \n",
       "1  {'uG2': 231.2313, 'arc': 2.124124, 'uG2Err': 2...  \n",
       "2  {'uG2': 231.2313, 'arc': 2.124124, 'uG2Err': 2...  \n",
       "3  {'uG2': 231.2313, 'arc': 2.124124, 'uG2Err': 2...  \n",
       "4  {'uG2': 231.2313, 'arc': 2.124124, 'uG2Err': 2...  "
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pdAlertsDf.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'ccdVisitId': 111111,\n",
       " 'decl': 0.126243049656,\n",
       " 'diaSourceId': 281323062375219200,\n",
       " 'filterName': 'r',\n",
       " 'flags': 0,\n",
       " 'midPointTai': 1480360995,\n",
       " 'psFlux': 1241.0,\n",
       " 'ra': 351.570546978,\n",
       " 'ra_decl_Cov': {'declSigma': 0.00028,\n",
       "  'raSigma': 0.00028,\n",
       "  'ra_decl_Cov': 0.00029},\n",
       " 'snr': 41.1,\n",
       " 'x': 112.1,\n",
       " 'x_y_Cov': {'xSigma': 1.2, 'x_y_Cov': 1.2, 'ySigma': 1.1},\n",
       " 'y': 121.1}"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pdAlertsDf['diaSource'][0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dict"
      ]
     },
     "execution_count": 52,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(pdAlertsDf['diaSource'][0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dict"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(pdAlertsDf['diaSource'][0]['ra_decl_Cov'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'declSigma': 0.00028, 'raSigma': 0.00028, 'ra_decl_Cov': 0.00029}"
      ]
     },
     "execution_count": 54,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pdAlertsDf['diaSource'][0]['ra_decl_Cov']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nested dicts look like they have survived, when creating a pandas dataframe from a list from a spark series."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "But pyspark.sql.dataframe creation can infer data structure incorrectly, if the data does not have a schema."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sparkdf = spark.createDataFrame([list(pdAlertsSeries)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 56,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(sparkdf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(_1={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _2={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _3={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _4={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _5={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _6={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _7={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _8={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _9={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _10={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _11={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _12={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _13={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _14={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _15={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _16={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _17={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _18={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _19={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None}, _20={'prv_diaSources': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'alertId': None, 'diaObject': None, 'l1dbId': None, 'diaSourcesL2': [{'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}, {'ra_decl_Cov': {'raSigma': 0.00028, 'ra_decl_Cov': 0.00029, 'declSigma': 0.00028}, 'x_y_Cov': {'ySigma': 1.1, 'x_y_Cov': 1.2, 'xSigma': 1.2}, 'diaSourceId': None, 'x': None, 'decl': None, 'snr': None, 'psFlux': None, 'y': None, 'filterName': None, 'flags': None, 'ra': None, 'midPointTai': None, 'ccdVisitId': None}], 'ssObject': None, 'diaObjectL2': None, 'diaSource': None})]"
      ]
     },
     "execution_count": 57,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sparkdf.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Summary\n",
    "\n",
    "Using Spark Structured Streaming with a Kafka formatted stream and Kafka stream values of alerts that are unstructured (non-Avro, strings) is possible for filtering, but really a roundabout solution, if you do either of the following:\n",
    "\n",
    "\n",
    "### Filter using list comprehension\n",
    "1. Read a structured stream from Kafka\n",
    "2. Convert \"values\" to strings.\n",
    "3. Construct a pyspark.sql.df selecting all of the values.\n",
    "4. Use rdd.map to do literal_eval on the strings to convert to rdds of dicts.\n",
    "5. Collect the rdds into a list of dicts.\n",
    "6. Filter on the list of dicts using list comprehension.\n",
    "\n",
    "But, issues can unknowingly arise if after step 4 you try and convert to pyspark.sql.dataframes to do filtering (using RDD.toDF() method).\n",
    "\n",
    "\n",
    "### Filter using pandas\n",
    "1. Read a structured stream from Kafka\n",
    "2. Convert \"values\" to strings.\n",
    "3. Construct a pyspark.sql.df selecting all of the values.\n",
    "4. Convert the above pyspark.sql.df toPandas().\n",
    "5. Column map \"values\" to do literal_eval on the strings to convert to a pandas series of dicts.\n",
    "6. Create a pandas dataframe from list(above series) and filter using pandas.\n",
    "\n",
    "But, again, issues can unknowingly arise if after step 5 you try and create a pyspark.sql.dataframe from the series of dicts to do filtering with pyspark.sql.dataframes.\n",
    "\n",
    "The value of using Spark Structured Streaming is primarily in the ability to use pyspark.sql on structured data, so for this example, using Spark Structured Streaming isn't particulary useful.\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}