There is another simple type of data profiling that we can also apply that helps with file inspection. It involves profiling the actual bytes that make up a whole file. It is an old method, one that originally comes from cryptography where frequency analysis of letters in texts was used to gain an edge on deciphering substitution codes.
While not a common technique in data science circles today, byte level analysis is surprisingly useful when it's needed. In the past, data encodings were a massive problem. Files were encoded in a range of code pages, across ASCII and EBCDIC standards. Byte frequency reporting was often critical to discover the actual encoding, delimiters, and line endings used in the files. Back, then the number of people who could create files, but not technically describe them, waqs surprising. Today, as the world moves increasingly to Unicode-based character encodings, these old methods need updating. In Unicode, the concept of a byte is modernized to multi-byte code points, which can be revealed in Scala using the following function:
val tst = "Andrew Introducing character class masks"
def toCodePointVector(input: String) = input.map{
case (i) if i > 65535 =>
val hchar = (i - 0x10000) / 0x400 + 0xD800
val lchar = (i - 0x10000) % 0x400 + 0xDC00
f"\\u$hchar%04x\\u$lchar%04x"
case (i) if i > 0 => f"\\u$i%04x"
// kudos to Ben Reich: http://k.bytefreq.com/1MjyvNz
}
val out = toCodePointVector(tst)
val rows = sc.parallelize(out)
rows.countByValue().foreach(println)
// results in the following: [codepoint], [Frequency_count]
(\u0065,1)
(\u03d6,1)
(\u006e,1)
(\u0072,1)
(\u0077,1)
(\u0041,1)
(\u0020,2)
(\u6f22,1)
(\u0064,1)
(\u5b57,1)
Using this function, we can begin to profile any international character level data we receive in our GDELT dataset and start to understand the complexities we might face in exploiting the data. But, unlike the other masks, to create interpretable results from code points, we require a dictionary that we can use to look up meaningful contextual information, such as unicode category and the unicode character names.
To generate a contextual lookup, we can use this quick command line hack to generate a reduced dictionary from the main one found at unicode.org, which should help us to better report on our findings:
# use "hdfs dfs -put" to load codepoints.txt to hdfs, so
# you can use it later
head -1300 codepoints.txt | tail -4
0513 Ll CYRILLIC SMALL
0514 Lu CYRILLIC CAPITAL
0515 Ll CYRILLIC SMALL
0516 Lu CYRILLIC CAPITAL
We will use this dictionary, joined to our discovered code points, to report on the character class frequencies of each byte in the file. While it seems like a simple form of analysis, the results can often be surprising and offer a forensic level of understanding of the data we are handling, its source, and the types of algorithms and methods we can apply successfully to it. We will also look up the general Unicode Category to simplify our reports using the following lookup table:
A simple but effective method for quickly exploring new types of data is to make use of mask based data profiling. A mask in this context is a transformation function for a string that generalizes a data item into a feature, that, as a collection of masks, will have a lower cardinality than the original values in the field of study.
When a column of data is summarized into mask frequency counts, a process commonly called data profiling, it can offer rapid insights into the common structures and content of the strings, and hence reveal how the raw data was encoded. Consider the following mask for exploring data:
Translate uppercase letters to A
Translate lowercase letters to a
Translate numbers, 0 through 9, to 9
It seems like a very simple transformation at first glance. As an example, let's apply this mask to a high cardinality field of data, such as the GDELT GKG file's V2.1 Source Common Name field. The documentation suggests it records the common name of the source of the news article being studied, which typically is the name of the website the news article was scraped from. Our expectation is that it contains domain names, such as nytimes.com.
Before implementing the production solution in Spark, let's prototype a profiler on the Unix command line to provide an example that we can run anywhere:
sed "s/[0-9]/9/g; s/[a-z]/a/g; s/[A-Z]/A/g" | sort | \
uniq -c | sort -r -n | head -20
232 aaaa.aaa
195 aaaaaaaaaa.aaa
186 aaaaaa.aaa
182 aaaaaaaa.aaa
168 aaaaaaa.aaa
167 aaaaaaaaaaaa.aaa
167 aaaaa.aaa
153 aaaaaaaaaaaaa.aaa
147 aaaaaaaaaaa.aaa
120 aaaaaaaaaaaaaa.aaa
The output is a sorted count of records found in the Source Common Name column alongside the mask generated by the regular expression (regex). It should be very clear looking at the results of this profiled data that the field contains domain names - or does it? As we have only looked at the most common masks (the top 20 in this case) perhaps the long tail of masks at the other end of the sorted list holds potential data quality issues at a lower frequency.
Rather than looking at just the top 20 masks, or even the bottom 20, we can introduce a subtle change to improve the generalization ability of our mask function. By making the regex collapse multiple adjacent occurrences of lower case letters into a single a character, the mask's cardinality can be reduced without really diminishing our ability to interpret the results. We can prototype this improvement with just a small change to our regex and hopefully view all the masks in one page of output:
$ # note: on a mac use gsed, on linux use sed.
$ hdfs dfs -cat 20150218230000.gkg.csv | \
gawk -F"\t" '{print $4}' | sed "s/[0-9]/9/g; s/[A-Z]/A/g; \
2 AAA Aa <---note here the pattern that stands out
2 9a99a.a
2 9a.a.a
1 a9.a.a
1 a.99a.a
1 9a9a.a
1 9999a.a
Very quickly, we have prototyped a mask that reduces the three thousand or so raw values down to a very short list of 22 values that are easily inspected by eye. As the long tail is now a much shorter tail, we can easily spot any possible outliers in this data field that could represent quality issues or special cases. This type of inspection, although manual, can be very powerful.
Notice, for instance, there is a particular mask in the output, AAA Aa, which doesn't have a dot within it, as we would expect in a domain name. We interpret this finding to mean we've spotted two rows of raw data that are not valid domain names, but perhaps general descriptors. Perhaps this is an error, or an example of what is known as, illogical field use, meaning there could be other values slipping into this column that perhaps should logically go elsewhere.
This is worth investigating, and it is easy to inspect those exact two records. We do so by generating the masks alongside the original data, then filtering on the offending mask to locate the original strings for manual inspection.
Rather than code a very long one liner on the command line, we can inspect these records using a legacy data profiler called bytefreq (short for byte frequencies) written in awk. It has switches to generate formatted reports, database ready metrics, and also a switch to output masks and data side by side. We have open-sourced bytefreq specifically for readers of this book, and suggest you play with it to really understand how useful this technique can be: https://bitbucket.org/bytesumo/bytefreq.
$ # here is a Low Granularity report from bytefreq
When we inspect the odd mask, A Aa, we can see the offending text found is BBC Monitoring, and in re-reading the GDELT documentation we will see that this is not an error, but a known special case. It means when using this field, we must remember to handle this special case. One way to handle it could be by including a correction rule to swap this string value for a value that works better, for example, the valid domain name www.monitor.bbc.co.uk, which is the data source to which the text string refers.
The idea we are introducing here is that a mask can be used as a key to retrieve offending records in particular fields. This logic leads us to the next major benefit of mask based profiling: the output masks are a form of Data Quality Error Code. These error codes can fall into two categories: a whitelist of good masks, and a blacklist of bad masks that are used to find poor quality data. Thought of this way, masks then form the basis for searching and retrieving data cleansing methods, or perhaps for throwing an alarm or rejecting a record.
The lesson is that we can create Treatment functions to remediate raw strings that are found using a particular mask calculated over data in a particular field. This thinking leads to the following conclusion: we can create a general framework around mask based profiling for doing data quality control and remediation as we read data within our data reading pipeline. This has some really advantageous solution properties:
Generating data quality masks is an on read process; we can accept new raw data and write it to disk then, on read, we can generate masks only when needed at query time - so data cleansing can be a dynamic process.
Treatment functions can then be dynamically applied to targeting remediation efforts that help to cleanse our data at the time of read.
Because previously unseen strings are generalized into masks, new strings can be flagged as having quality issues even if that exact string has never been seen before. This generality helps us to reduce complexity, simplify our processes, and create reusable smart solutions - even across subject areas.
Data items that create masks that do not fall either into mask white-lists, fix-lists, or blacklists can potentially be quarantined for attention; human analysts can inspect the records and hopefully whitelist them, or perhaps create new Treatments Functions that help to get the data out of quarantine and back into production.
Data quarantines can be implemented simply as an on-read filter, and when new remediation functions are created to cleanse or fix data, the dynamic treatments applied at read time will automatically release the corrected data to users without long delays.
Eventually a data quality Treatment library will be created that stabilizes over time. New work is mainly done by mapping and applying the existing treatments to new data. A phone number reformatting Treatment function, for example, can be widely reused over many datasets and projects.
With the method and architectural benefits now explained, the requirements for building a generalized mask based profiler should be clearer. Note that the mask generation process is a classic Hadoop MapReduce process: map input's data out to masks, and reduce those masks back down to summarized frequency counts. Note also how, even in this short example, we have already used two types of masks and each is made up of a pipeline of underlying transformations. It suggests we need a tool that supports a library of predefined masks as well as allowing for user defined masks that can be created quickly and on demand. It also suggests there should be ways to stack the masks to build them up into complex pipelines.
What may not be so obvious yet is that all data profiling done in this way can write profiler metrics to a common output format. This helps to improve reusability of our code through simplifying the logging, storing, retrieval, and consumption of the profiling data.
As an example we should be able to report all mask based profiler metrics using the following schema:
Metric Descriptor
Source Studied
IngestTime
MaskType
FieldName
Occurrence Count
KeyCount
MaskCount
Description
Once our metrics are captured in this single schema format, we can then build secondary reports using a user interface, such as Zeppelin notebook.
Before we walk through implementing these functions, an introduction to the characte
The next stage is to obtain the GDELT data and load it into GeoMesa. There are a number of options here, depending upon how you plan to proceed; if you are just working through this chapter, then you can use a script to download the data in one go:
$ for file in `cat md5sums | cut -d' ' -f3 | grep '^201[56]'` ; do wget http://data.gdeltproject.org/events/$file ; done
$ md5sum -c md5sums 2>&1 | grep '^201[56]'
This will download and verify all of the GDELT events data for 2015 and 2016. The amount of data required is something we need to estimate at this stage, as we do not know how our algorithm is going to work out, so we have chosen two years worth to start with.
An alternative to the script is to read Chapter 2, Data Acquisition, which explains in detail how to configure Apache NiFi to download the GDELT data in real time, and further it loads it to HDFS ready for use. Otherwise, a script to allow the preceding data to be transferred to HDFS is shown as follows:
HDFS uses data blocks; we want to ensure that files are stored as efficiently as possible. Writing a method to aggregate files to the HDFS block size (64 MB by default) will ensure the NameNode memory is not filled with many entries for lots of small files, and will make processing more efficient also. Large files that use more than one block (file size > 64 MB) are known as split files.
We have a substantial amount of data in HDFS (approximately 48 GB for 2015/16). Now, we will load this to Accumulo via GeoMesa.
Since MapReduce (MR) is generally considered dead, or at least dying, it is very useful to know how to create Spark jobs from those existing in MR. The following method can be applied to any MR job. We will consider the GeoMesa Accumulo loading job described in the GeoMesa tutorial (geomesa-examples-gdelt) for this case.
An MR job is typically made up of three parts: the mapper, the reducer, and the driver. The GeoMesa example is a map-only job and therefore requires no reducer. The job takes a GDELT input line, creates a (Key,Value) pair from an empty Text object and the created GeoMesa SimpleFeature, and uses the GeoMesaOutputFormat to load the data to Accumulo. The full code of the MR job can be found in our repository; next this we will work through the key parts and suggest the changes required for Spark.
The job is initiated from the main method; the first few lines are related to parsing the required options from the command line, such as the Accumulo username and password. We then reach:
The GeoMesa SimpleFeatureType is the primary mechanism used to store data in a GeoMesa data store and it needs to be initialized once, along with the data store initialization. Once this is done we execute the MR job itself. In Spark, we can pass the arguments via the command line as before, and then do the one-off setup:
spark-submit --class io.gzet.geomesa.ingest /
--master yarn /
geomesa-ingest.jar <accumulo-instance-id>
...
The contents of the jar contain a standard Spark job:
val conf = new SparkConf()
val sc = new SparkContext(conf.setAppName("Geomesa Ingest"))
Parse the command line arguments as before, as well as performing the initialization:
val featureType = buildGDELTFeatureType(featureName)
val ds = DataStoreFinder
.getDataStore(dsConf)
.createSchema(featureType)
Now we can load the data from HDFS, using wildcards if required. This creates one partition for each block of the file (64 MB default), resulting in an RDD[String]:
val distDataRDD = sc.textFile(/data/gdelt/*.CSV)
Or we can fix the number of partitions, depending upon our available resources:
val distDataRDD = sc.textFile(/data/gdelt/*.CSV, 20)
Then we can perform the map, where we can embed the function to replace the process in the original MR map method. We create a tuple (Text,SimpleFeatureType) to replicate a (Key, Value) pair so that we can use the OutputFormat in the next step. When Scala Tuples are created in this way, the resulting RDD gains extra methods, such as ReduceByKey, which is functionally equivalent to the MR Reducer (see below for further information on what we should really be using, mapPartitions):
val processedRDD = distDataRDD.map(s =>{
// Processing as before to build the SimpleFeatureType
(new Text, simpleFeatureType)
})
Then, we can finally output to Accumulo using the GeomesaOutputFormat from the original job:
At this stage, we have not mentioned the setup method in the MR job; this method is called before any input is processed to allocate an expensive resource like a database connection, or in our case, a reusable object, and a cleanup method is then used to release that resource if it were to persist when out of scope. In our case, the setup method is used to create a SimpleFeatureBuilder which can be reused during each call of the mapper to build SimpleFeatures for output; there is no cleanup method as the memory is automatically released when the object is out of scope (the code has completed).
The Spark map function only operates on one input at a time, and provides no means to execute code before or after transforming a batch of values. It looks reasonable to simply put the setup and cleanup code before and after a call to map:
// do setup work
val processedRDD = distDataRDD.map(s =>{
// Processing as before to build the SimpleFeatureType
(new Text, simpleFeatureType)
})
// do cleanup work
But, this fails for several reasons:
It puts any objects used in map into the map function's closure, which requires that it be serializable (for example, by implementing java.io.Serializable). Not all objects will be serializable, thus exceptions may be thrown.
The map function is a transformation, rather than an operation, and is lazily evaluated. Thus, instructions after the map function are not guaranteed to be executed immediately.
Even if the preceding issues were covered for a particular implementation, we would only be executing code on the driver, not necessarily freeing resources allocated by serialized copies.
The closest counterpart to a mapper in Spark is the mapPartitions method. This method does not map just one value to another value, but maps an Iterator of values to an Iterator of other values, akin to a bulk-map method. This means that the mapPartitions can allocate resources locally at its start:
val processedRDD = distDataRDD.mapPartitions { valueIterator =>
// setup code for SimpleFeatureBuilder
val transformed = valueIterator.map( . . . )
transformed
}
However, releasing resources (cleanup) is not straightforward as we still experience the lazy evaluation problem; if resources are freed after the map, then the iterator may not have evaluated before the disappearance of those resources. One solution to this is as follows:
val processedRDD = distDataRDD.mapPartitions { valueIterator =>
if (valueIterator.isEmpty) {
// return an Iterator
} else {
// setup code for SimpleFeatureBuilder
valueIterator.map { s =>
// Processing as before to build the SimpleFeatureType
加好友,备注jltj
京公网安备 11010802022788号
论坛法律顾问：王进律师
知识产权保护声明
免责及隐私声明
发帖
雷达卡
显身卡
