SS.ipynb

{"cells":[{"cell_type":"code","source":["\nfrom pyspark.sql.types import *\nfrom pyspark.sql.functions import *\n\n# Since we know the data format already, let's define the schema to speed up processing (no need for Spark to infer schema)\njsonSchema = StructType([ StructField(\"time\", TimestampType(), True), StructField(\"action\", StringType(), True) ])\n\ninputPath = \"s3://kidddcorp-sparkdata/events/\"\n# Similar to definition of staticInputDF above, just using `readStream` instead of `read`\nstreamingInputDF = (\n  spark\n    .readStream                       \n    .schema(jsonSchema)               # Set the schema of the JSON data\n    .option(\"maxFilesPerTrigger\", 1)  # Treat a sequence of files as a stream by picking one file at a time\n    .json(inputPath)\n)\n\n# Same query as staticInputDF\nstreamingCountsDF = (                 \n  streamingInputDF\n    .groupBy(\n      streamingInputDF.action, \n      window(streamingInputDF.time, \"1 hour\"))\n    .count()\n)\n\n# Is this DF actually a streaming DF?\nstreamingCountsDF.isStreaming\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")  # keep the size of shuffles small\n\nquery = (\n  streamingCountsDF\n    .writeStream\n    .format(\"memory\")        # memory = store in-memory table \n    .queryName(\"counts\")     # counts = name of the in-memory table\n    .outputMode(\"complete\")  # complete = all the counts should be in the table\n    .start()\n)"],"metadata":{"application/vnd.databricks.v1+cell":{"title":"","showTitle":false,"inputWidgets":{},"nuid":"baf98274-bd5c-4081-9521-6dca77b44da6"}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[],"data":"<div class=\"ansiout\"></div>","removedWidgets":[],"addedWidgets":{},"metadata":{},"type":"html","arguments":{}}},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"]}}],"execution_count":0},{"cell_type":"code","source":[""],"metadata":{"application/vnd.databricks.v1+cell":{"title":"","showTitle":false,"inputWidgets":{},"nuid":"aa72797f-cb11-4d46-8d91-ec5a94ee2296"}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"data":"","errorSummary":"","metadata":{},"errorTraceType":null,"type":"ipynbError","arguments":{}}},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>"]}}],"execution_count":0}],"metadata":{"application/vnd.databricks.v1+notebook":{"notebookName":"SS","dashboards":[],"notebookMetadata":{"pythonIndentUnit":4},"language":"python","widgets":{},"notebookOrigID":2786165303007475}},"nbformat":4,"nbformat_minor":0}


#%sql select action, date_format(window.end, "MMM-dd HH:mm") as time, count from counts order by time, action