GCDEC/Dataflow/Notes - Revision history

Admin: /* Resources */

2017-10-17T23:56:53Z

Resources

@@ Line 1,421: / Line 1,421: @@
 * Cancel - stops the pipeline immediately
 * Drain - finishes processing the rest of the contents of the PubSub stream
 =Resources=
 =Flags=

Admin: /* Streaming Lab */

2017-10-17T23:55:31Z

Streaming Lab

← Older revision		Revision as of 23:55, 17 October 2017
Line 1,342:		Line 1,342:

	===Streaming Lab===		===Streaming Lab===

			Link to lab: https://codelabs.developers.google.com/codelabs/cpb101-bigquery-dataflow-streaming/

			Objective:
			* Create and read from a PubSub topic
			* Use Dataflow to aggregate streaming records coming in (real-time) via PubSub (using windowed aggregate)
			* Stream aggregate statistics into BigQuery
			* Analyze results while data are streaming into data warehouse


			====Set Up PubSub and BigQuery====

			Start by creating BigQuery data set called "demo"

			Then create PubSub data source called "streamdemo"

			Back to the same training-data-analyst repository: https://github.com/GoogleCloudPlatform/training-data-analyst

			View the pipeline code at StreamDemoConsumer.java: https://github.com/GoogleCloudPlatform/training-data-analyst/blob/b867c7039fafbe9616b1a6304fc74d14159a05a5/courses/data_analysis/lab2/javahelp/src/main/java/com/google/cloud/training/dataanalyst/javahelp/StreamDemoConsumer.java

			====The Java PubSub Consumer Code====

			Overview of StreamDemoConsumer code:

			Main method does the following:
			* Create pipeline options factory to extract command line arguments
			* Build table schema for the sink of this pipeline (timestamps, number of words)
			* Create the pipeline parts:
			* "GetMessages" function reads strings from the specified topic
			* "Window" applies a sliding window (2 minute window, every 30 seconds)
			* "WordsPerLine" (parallel function) splits String line and counts number of words
			* "WordsInTimeWindow" applies Sum.integersGlobally() aggregate function
			* "ToBQRow" (parallel function) creates a TableRow object and sets the two column values, "timestamp" and "num_words"
			* Finally, we apply BigQuery.writeTableRow(), TO the output, WITH our schema
			* Last but not least, we actually run the pipeline.

			To run, use the run_oncloud4.sh script, which hard-codes the name of the class.

			IMPORTANT: this pipeline does not automatically exit - when you run the script, it will build the pipeline, submit the job, and will report a build success. It will then return you back to the command prompt.

			To generate data for the PubSub topic: open the PubSub section of the Cloud Console, and look at the top for "Publish Message"

			(I published a few messages)

			Now go to BigQuery and see if you can query the data warehouse. I was not able to. No tables had showed up yet, no fields.

			(Switched to Dataflow section of Cloud Console, saw that the graph was still being analyzed)

			Waited a bit, switched back to BigQuery:

			<pre>
			SELECT timestamp, num_words
			FROM [not-all-broken:demos.streamdemo]
			ORDER BY timestamp ASC
			</pre>

			Bingo, some results came up:

			<pre>
			timestamp,num_words
			2017-10-17 00:09:08.267999 UTC,391
			2017-10-17 00:09:36.210999 UTC,486
			2017-10-17 00:10:07.535000 UTC,486
			2017-10-17 00:10:35.854000 UTC,486
			2017-10-17 00:11:08.479000 UTC,270
			2017-10-17 00:11:35.796999 UTC,175
			2017-10-17 00:12:08.414000 UTC,335
			2017-10-17 00:12:35.891000 UTC,585
			2017-10-17 00:13:08.217000 UTC,410
			2017-10-17 00:13:35.536000 UTC,410
			2017-10-17 00:14:07.852999 UTC,250
			2017-10-17 00:15:07.492000 UTC,38
			</pre>

			To stop the pipeline, use the Cloud SDK command line tool or the Cloud Console.

			Open the job, click "Stop Job" on the right side.
			* Cancel - stops the pipeline immediately
			* Drain - finishes processing the rest of the contents of the PubSub stream

	=Resources=		=Resources=

Admin: /* Streaming Data into Dataflow */

2017-10-17T23:55:15Z

Streaming Data into Dataflow

← Older revision		Revision as of 23:55, 17 October 2017
Line 1,298:		Line 1,298:

	===Streaming Data into Dataflow===		===Streaming Data into Dataflow===

			Serverless data analysis and streaming

			Previous labs: we have been running data using a batch workflow (reading stored data)

			But we can apply same workflows/pipelines to streaming data sources (e.g., PubSub)

			Dataflow handles out-of-order records, delays, etc. that happen with streaming sources

			Can associate timestamp with inputs:
			* PubSub has automatic timestmap
			* Timestamp is time that message was published to topic
			* For batch, we explicitly assign a timestamp wherever we might need one

			Example:

			<pre>
			c.outputWithTimestamp(f, Instant.parse(fields[2]));
			</pre>

			This allows you to "trick" your streaming pipeline into handling batch data as well

			Use windows for aggregating unbounded ("infinite") collections

			Example: sliding window of 2 minutes, every 30 seconds

			<pre>
			.apply("window", Window.into( SlidingWindows
			.of(Duration.standardMinutes(2))
			.every(Duration.standardSeconds(30))
			))
			</pre>

			Then when you do a group by key, group sums, aggregations, etc. - all carried out in the context of that sliding 2 minute window

			Doing a mean-per-key is now no longer a mean per key over ENTIRE data set, it is now over the sliding window ONLY

			Think of PubSub as the shock absorber - it absorbs all the data

			That's then passed on to Dataflow, and added to BigQuery

			You can still query BigQuery data warehouse, even as data are streaming in

	===Streaming Lab===		===Streaming Lab===

Admin: /* Side Inputs Lab */

2017-10-17T23:44:34Z

Side Inputs Lab

Show changes

Admin: /* Side Inputs */

2017-10-17T23:35:38Z

Side Inputs

← Older revision		Revision as of 23:35, 17 October 2017
Line 1,052:		Line 1,052:

	===Side Inputs===		===Side Inputs===

			Side inputs and in-memory objects:

			We talked about a canonical form of processing in a data pipeline - we read data in from a source, we do some transforms, we write data out to a sink.

			However, in reality we often need additional information during the transform process. Maybe you need external data - don't have all the information you need to process the tax form.

			Suppose you have in-memory object that will not change - something fixed. Those can be provided as constructor parameters to each transform. Example: when we do the grep transform, we can say, "make the search term a String parameter."

			<pre>
			public class Match extends DoFn<String, String> {

			private final String searchTerm;

			public Match(String searchTerm) {
			this.searchTerm = searchTerm;
			}

			@Override
			public void processElement(ProcessContext c) throws Exception {
			String line = c.element();
			if(line.contains(searchTerm)) {
			c.output(line);
			}
			}
			}
			</pre>

			And later in the code, when we are actually assembling the pipeline, we can provide the grep term at the time of pipeline assembly:

			<pre>
			p.apply("Grep", ParDo.of( new Match(searchTerm) ) );
			</pre>

			This is not calculated dynamically - it's just provided, it's fixed in memory.

			But SOMETIMES, you need an entirely separate PCollection (a second data source) to process the data from one source (one PCollection).

			If you need a second PCollection, the procedure is as follows:
			* Take the other source (the smaller of the two) and convert it into a view
			* Can either be a list or a map - list if PCollection of objects, map if PCollection of key-value pairs
			* We turn the PCollection into a PCollectionView
			* This can now be provided as a side input
			* The main input is the first PCollection
			* The side input is the second PCollection - pass the view of that PCollection in as a side input
			* ParDo.of becomes ParDo.withSideInputs

			Inside of the ParDo that has the side input, you can now access the side inputs via the process context, c.sideInput()

			<pre>
			// Convert PCollection to a View (asList, asMap)
			PCollection<KV<String, Integer>> cz = ...
			PCollectionView<Map<String, Integer>> czmap = cz.apply("ToView", View.asMap());

			// Call the ParDo with side input[s]
			.apply("...", ParDo.withSideInputs(czmap)
			.of(new DoFn<KV<String, Integer>>, KV<String, Double>>()
			))

			// Within the ParDo, get the side input from the process context
			public void processElement(ProcessContext c) throws Exception {
			Integer fromcz = c.sideInput(czmap).get(czkey);
			// This uses .get() because the side input is a map.
			}
			</pre>

			So the crucial components are:
			* Converting PCollection to PCollectionView
			* Pass the PCollectionView as a side input
			* Once that's done, you can get the side input from the process context, and use it just like a regular map or list

	===Side Inputs Lab===		===Side Inputs Lab===

Admin: /* MapReduce Lab */

2017-10-17T23:21:01Z

MapReduce Lab

Show changes

Admin: /* MapReduce Transforms */

2017-10-17T23:20:37Z

MapReduce Transforms

← Older revision		Revision as of 23:20, 17 October 2017
Line 698:		Line 698:
	===MapReduce Transforms===		===MapReduce Transforms===

	Now, let's cover how these map operations actually look.		Now, let's cover how these map operations actually look. There are two basic kinds of mapping operations:
			* One-to-one map operations
			* One-to-many map operations

			These are treated differently in Python versus Java.

	====Python Transforms====		====Python Transforms====
Line 704:		Line 708:
	Python:		Python:
	* Can use a Map or a FlatMap		* Can use a Map or a FlatMap
	* Map - ~~1:1~~ relationship between input and output (every input will result in some kind of output)		* Map - one-to-one relationship between input and output (every input will result in some kind of output)
	* FlatMap - ~~not a 1:1~~ relationship between input and output (usually with generator)		* FlatMap - one-to-many relationship between input and output (usually with generator)
	* Generators or filters - we are choosing whether to return something or not; we may end up returning many things		* Generators or filters - we are choosing whether to return something or not; we may end up returning many things

Line 727:		Line 731:

	Java:		Java:
	* In both cases, use apply (with a ParDo) for both cases		* In both cases (one-to-one or one-to-many), use apply (with a ParDo) for both cases

	We already covered an example of a flat map (returning zero, one, or multiple outputs for a single input).		We already covered an example of a flat map (returning zero, one, or multiple outputs for a single input).

Admin: /* MapReduce with Dataflow */

2017-10-17T23:19:28Z

MapReduce with Dataflow

Show changes

Admin: /* MapReduce with Dataflow */

2017-10-17T23:19:03Z

MapReduce with Dataflow

← Older revision		Revision as of 23:19, 17 October 2017
Line 662:		Line 662:

	===MapReduce with Dataflow===		===MapReduce with Dataflow===

			When we take the MapReduce idea and apply Dataflow to it:
			* Data is split into chunks
			* Each chunk is sent to a node
			* The request, or job, contains "map" instructions
			* Maps are sent to each node holding data
			* The map is executed on the data on the node
			* Finally, the reduce operation gathers/combines the data onto a single node

			Map operations - highly parallelizable

			Reduce operations - computations performed on a large number of rows

			Example: each node holds a tax return, computes a return amount, then the reduce operation is accumulating state-by-state statistics

			Tax return example:
			* Data: all tax returns
			* Each node: processes a few tax returns
			* Each tax return: processed by map node
			* Reduce operation: computing a marginal tax rate
			* Reduce nodes may process one or two states

			Dataflow: ParDo (parallel do)
			* ParDo acts on one item at a time
			* Similar to Map in MapReduce
			* Multiple instances of a ParDo class on many machines, so should be stateless
			* That is, map operations operate statically on the data, do not "remember" things
			* Aggregation (e.g., time-average) should be done by reducer nodes, not compute nodes
			* Useful for certain kinds of operations:
			* Filtering (choosing which inputs to emit for the next step of the pipeline/computation)
			* Converting one type to another (parallel collection of floats to parallel collection of strings)
			* Extracting parts of an input (one particular field of a table row, e.g., extract Box 17 from a form)
			* Combining input values in a calculation (combine different parts of a document to calculate a percentage of taxes)

	===MapReduce Lab===		===MapReduce Lab===

Admin: /* Serverless Data Pipeline in Python */

2017-10-17T23:18:20Z

Serverless Data Pipeline in Python

← Older revision		Revision as of 23:18, 17 October 2017
Line 524:		Line 524:
	====Serverless Data Pipeline in Python====		====Serverless Data Pipeline in Python====

			Link to Python lab files: https://github.com/GoogleCloudPlatform/training-data-analyst/blob/master/courses/data_analysis/lab2/python/install_packages.sh

	Start by running the install script to update pip:		Start by running the install script to update pip: