增强现实AR>> facets-overview-spark>> 返回
项目作者： gopro

项目描述 ：
Spark Implementation of Google Facets Overview https://github.com/PAIR-code/facets
高级语言： Scala
项目地址: git://github.com/gopro/facets-overview-spark.git
创建时间： 2018-05-02T18:48:31Z
项目社区：https://github.com/gopro/facets-overview-spark
开源协议：Other
下载


Facets Overview Spark

What is “Facets Overview Spark”

Facets Overview Spark is an implementation of Google Facet Overview (https://github.com/PAIR-code/facets) using Spark and Scala.

The project is open sourced by GoPro Data Science Engineering team.

Introduction

Google Open Sourced the Facets Project in 2017 (https://github.com/PAIR-code/facets), which can help Data Scientists to better understand the data set
and under the slogan that “Better data leads to better models”.

The Google “Facets” includes two sub-projects: Facets-overview and Facets-dive.

“Facets Overview “takes input feature data from any number of datasets, analyzes them feature by feature and visualizes the analysis.

Based on Facets( github page: https://github.com/PAIR-code/facets)

  1. Overview gives a high-level view of one or more data sets. It produces a visual feature-by-feature statistical analysis, and can also be used to compare statistics across two or more data sets. The tool can process both numeric and string features, including multiple instances of a number or string per feature.
  2. Overview can help uncover issues with datasets, including the following:
  4. * Unexpected feature values
  5. * Missing feature values for a large number of examples
  6. * Training/serving skew
  7. * Training/test/validation set skew
  8. Key aspects of the visualization are outlier detection and distribution comparison across multiple datasets. Interesting values (such as a high proportion of missing data, or very different distributions of a feature across multiple datasets) are highlighted in red. Features can be sorted by values of interest such as the number of missing values or the skew between the different datasets.

Currently Google is no longer active develop Facets Project, but the Facets-overview has been integrated into TFX Validation (https://www.tensorflow.org/tfx/data_validation/get_started, https://github.com/tensorflow/tfx/blob/master/docs/guide/statsgen.md)
and What-If-Tools (https://pair-code.github.io/what-if-tool/). In TFX, one can launch the facets-overview visualization like

  1.     tfdv.visualize_statistics(stats)

With Facets-overview-spark, not only you can generate the statistics using spark for big data set, but also you can visualize
the statistics with Facets just like TFX validation, with simple HTML or jupyter notebook (with out depends on TFX)

Facets-overview Implementations

The Facets-overview is consists of

  • Feature Statistics Protocol Buffer
  • Feature Statistics Generation
  • Visualization

The implementation of Feature Statistics Generation have two flavors : Python (used by Jupyter notebook) and Javascripts (used by web )

Overview visualization

Current implemention is in python depends on Numpy or Javascripts. Which means the data statics generation is limited by
one machine or one browser.

This project provides an additional implementation for Statistics generation.
We can use Spark to leverage the spark generate stats with distributed computinig capability

Overview visualization with Spark

Design Considerations

  • We need to find a Scala Protobuf generator
  • The python implementation is in loop-mutable and update fasion, we need re-arrange it to use immutable collection
  • Numpy Implementation load all the data is in Array on the current computer, we need some way to transform data without load all data into memory.
  • We also need to find equivalent Numpy functions such as avg, mean, stddev, histogram
  • The “overview” implementation can be apply to Tensorflow Record (tf.sampels, tf.sequenceSamples), where the data value could be array or array of array of data.
  • In this implementation, we can leveraging tensorflow/ecosystem/spark-tensorflow-connector for tensorflow record support, where we can load the TFRecords into spark DataFrame. Once the DataFrame is created, rest of implementation is almost same.
  • We use DataFrame to represent the Feature. this is equivalent the feature array used in the Numpy.
  • Efficiency is not the major concern in this first version of design, we may have to pass data in multiple passes.
  • Feature names need to be santized (replace white space, dash (-) or special characters to underscore)

Data Structures

Based on Feature_statistics Protobuf definitions,

  • Each dataset has DatasetFeatureStatistics which contains lists metadata (name of dataset, number of sample, and Seq[FeatureNameStatistics])
  • Each FeatureNameStatistics is feature stats for the given dataset, which includes feature name and data type and one of (NumericStatistics,StringStatistics or binary_stats)
  • NumericStatistics consists CommonStatistics (std_dev,mean, median,min, max, histogram)
  • StringStatistics consists CommonStatistics as well as (unique values, {value, frequency}, top_values,avg_length, rank_histogram )

Google’s Python implementation mainly use dictionary to hold data structures. In this implemenation, we define several additional data structures to help organize the data.

  • Each dataset can be presented by the NamedDataFrame
  • Each dataset can be split with different features, so that it can also be converted to DataEntrySet
  • Each DataEntry represents one feature with basic information as well the DataFrame for the feature.
  • Each Feature associate with FeatureNameStatistics defined above, we use BasicNumStats and BasicStringStats to capture the basic statistics
  1. case class NamedDataFrame(name:String, data: DataFrame)
  2. case class DataEntrySet(name: String, size: Long, entries : Array[DataEntry])
  3. case class DataEntry(featureName: String,
  4.                      `type` : ProtoDataType,
  5.                      values:DataFrame,
  6.                      counts: DataFrame,
  7.                      missing : Long,
  8.                      feat_lens: Option[DataFrame] = None
  9.                     )
  10. case class BasicNumStats(name: String,
  11.                          numCount: Long = 0L,
  12.                          numNan :Long = 0L,
  13.                          numZeros:Long = 0L,
  14.                          numPosinf:Long = 0,
  15.                          numNeginf: Long = 0,
  16.                          stddev : Double = 0.0,
  17.                          mean   : Double = 0.0,
  18.                          min    : Double = 0.0,
  19.                          median : Double = 0.0,
  20.                          max    : Double = 0.0,
  21.                          histogram: (Array[Double], Array[Long])
  22.                         )
  24. case class BasicStringStats(name: String,
  25.                             numCount: Long = 0,
  26.                             numNan :Long = 0L
  27.  )

Main class

  1. class FeatureStatsGenerator(datasetProto: DatasetFeatureStatisticsList) {
  3.  def protoFromDataFrames(dataFrames: List[NamedDataFrame],
  4.                          features : Set[String] = Set.empty[String],
  5.                          histgmCatLevelsCount:Option[Int]=None): DatasetFeatureStatisticsList = ???
  8. }

notice here that DatasetFeatureStatisticsList is class generated based on protobuf definition

FeatureStatsGenerator

Usage Samples

Generate Protobuf from DataFrame (using CSV)

In this example, load the CSV file (adult.data.csv, adult.test.csv) into Spark DataFrame
then pass the DataFrame to FeatureStatsGeneator to generate the protobuf class.

The file can then be persisted into a binary protobuf file, or based64 encoded binary protobuf file.

Few utility functions for loading from CSV and persist to file are provided

As you can see, once the DataFrame is created, the rest of the code is the same.

The examples can easily used for DataFrames from JSON, SQL, Tensorflow Records etc.

For instances, I can simply change the functions to

  1.     val trainData: DataFrame = loadJSONFile("path/to/data/jsonfile")
  2.     val testData : DataFrame  = loadJSONFile("path/to/test/jsonfile")
  4.     val trainData: DataFrame = sqlContext.sql("select * from DataTable")
  5.     val testData : DataFrame = sqlContext.sql("select * from TestTable")

Here is the example:

First load CSV files (data and test) into DataFrames

  1.     val features = Array("Age", "Workclass", "fnlwgt", "Education", "Education-Num", "Marital Status",
  2.                          "Occupation", "Relationship", "Race", "Sex", "Capital Gain", "Capital Loss",
  3.                          "Hours per week", "Country", "Target")
  5.     val trainData: DataFrame = loadCSVFile("src/test/resources/data/adult.data.csv")
  6.     val testData : DataFrame = loadCSVFile("src/test/resources/data/adult.test.txt")
  1. Next Associate the column names ("schema") to the loaded DataFrame, then created a
  2. list of "Named DataFrame"
  1.     val spark = sqlContext.sparkSession
  2.     import spark.implicits._
  4.     val train = trainData.toDF(features: _*)
  5.     val test = testData.toDF(features: _*)
  7.     val dataframes = List(NamedDataFrame(name = "train", train), 
  8.                           NamedDataFrame(name = "test", test))

Next, we create FeatureStatsGenerator, and passed the namedDataFrame list to the generator,
then call protoFromDataFrames(dataframes) to generate the stats.

  1.     val generator = new FeatureStatsGenerator(DatasetFeatureStatisticsList())
  2.     val proto = generator.protoFromDataFrames(dataframes)
  1. Once we have the feature stats probuf, we can save it to file, which can be loaded into web
  2. or jupyter notebook.
  1.     persistProto(proto,base64Encode = false, new File("src/test/resources/data/stats.pb"))
  2.     persistProto(proto,base64Encode = true, new File("src/test/resources/data/stats.txt"))

Here are some utility functions

  2.   private def loadCSVFile(filePath: String) : DataFrame = {
  3.     val spark = sqlContext.sparkSession
  4.     spark.read
  5.       .format("org.apache.spark.sql.execution.datasources.csv.CSVFileFormat")
  6.       .option("header", "false") //reading the headers
  7.       .option("mode", "DROPMALFORMED")
  8.       .option("inferSchema", "true")
  9.       .load(filePath)
  10.   }
  13.   private def persistProto(proto: DatasetFeatureStatisticsList, base64Encode: Boolean = false, file: File ) = {
  14.     if (base64Encode) {
  15.       import java.util.Base64
  16.       val b = Base64.getEncoder.encode(proto.toByteArray)
  17.       import java.nio.charset.Charset
  18.       import java.nio.file.{Files, Paths}
  19.       val  UTF8_CHARSET = Charset.forName("UTF-8")
  21.       Files.write(Paths.get(file.getPath), new String(b, UTF8_CHARSET).getBytes())
  22.     }
  23.     else {
  24.       Files.write(Paths.get(file.getPath), proto.toByteArray)
  25.     }
  26.   }

Generating stats for SequenceExample data

  1. In this example, we generate feature stats for the DataFrame
  2. that is the same as TFRecord SequenceExample
  4. The protobuf is then print to Json for easy to read.
  1.      val data = Seq((Seq(Seq("Tim Robbins","Morgan Freeman"), Seq("Brad Pitt","Edward Norton","Helena Bonham Carter")),
  2.                         Seq(Seq("The Shawshank Redemption"),Seq("Fight Club")),
  3.                         Seq(Seq(9.0), Seq(9.7)),
  4.                         19L,
  5.                         List("Majesty Rose", "Savannah Outen", "One Direction"),
  6.                         "pt_BR"
  7.                       ))
  8.         val featureNames = Seq("Movie Actors" ,"Movie Names",  "Movie Ratings","Age" ,"Favorites",  "Locale")
  9.         val dataDF = spark.createDataFrame(data).toDF(featureNames:_*)
  12.         dataDF.printSchema()
  13.         dataDF.show()
  14.         //generate datastats
  15.         val dataframes = List(NamedDataFrame(name = "data", dataDF))
  16.         val p = generator.protoFromDataFrames(dataframes)
  18.         println("json=" + toJson(p))
  21.   private def toJson(proto: DatasetFeatureStatisticsList) : String = {
  22.     import scalapb.json4s.JsonFormat
  23.     JsonFormat.toJsonString(proto)
  24.   }

Generate Stats for TFRecords

  1.  val spark = sqlContext.sparkSession
  3.  val schema = StructType(List(StructField("id", IntegerType),
  4.               StructField("IntegerTypeLabel", IntegerType),
  5.               StructField("LongTypeLabel", LongType),
  6.               StructField("FloatTypeLabel", FloatType),
  7.               StructField("DoubleTypeLabel", DoubleType),
  8.               StructField("VectorLabel", ArrayType(DoubleType, containsNull =  true)),
  9.               StructField("name", StringType)))
  12.  val df =  TFRecordHelper.loadTFRecords(spark, path,  TFRecordType.Example, Some(schema))
  14.  val p  = generator.protoFromDataFrames(dataframes)
  16.  println("json=" + toJson(p))

The TFRecordHelper is small utility class which allow your to use leverage enumerated TFRecordType
and pass-in optional schema

How to use the generated feature statistics in Jupyter notebook (scala, python)

This demo simulates the original facets-overview with
(https://github.com/PAIR-code/facets/blob/master/facets_overview/Overview_demo.ipynb)
in the original demo train data and test data are located at

https://archive.ics.uci.edu/ml/machine-learning-databases/adult/adult.data
https://archive.ics.uci.edu/ml/machine-learning-databases/adult/adult.test

we decided to pre-download the data, the data is not large, so we store data in the project
and we can work offline.

the original demo has the following parts (in different cells, in Jupyter Notebook)

  • Part 1: Generate Panda DataFrames from CSV files
  • Part 2: Pass the data frames to the features stats generator to generate protobuf
  • Part 3: Base64 Encode the Protobuf structure to String

  • Part 4: Pass the probuf string to HTML Display

    We are going to do the same

  • Part 1: Use Spark to generate Spark DataFrames from CSV files (Scala)

  • Part 2: Pass the data frames to the features stats generator to generate protobuf (Scala + Spark), probuf is save to local file
  • Part 3 In Jupyter Notebook, we read the protobuf from file and encoded to base64 if needed.
  • Part 4: Pass the probuf structure to HTML Display

Let’s dive a bit more details on each part.

  • Part 1: Generate Spark Data Frames
  1.     //original feature columns
  2.     val columns = Array("Age", "Workclass", "fnlwgt", "Education", "Education-Num", "Marital Status",
  3.       "Occupation", "Relationship", "Race", "Sex", "Capital Gain", "Capital Loss",
  4.       "Hours per week", "Country", "Target")
  6.     //normalize the feature names by removing dash, space etc.
  7.     val features: Array[String] = DataFrameUtils.sanitizedNames(columns)
  1. Slightly different from the original examaple, we sanitized the column, Column contains "-" or white space ' ' 
  2. or special characters may not be works well if one decide to persist the data frame to hive table.
  4. The libary always santize the column name during stats generation as part of the design consideration
  5. so we need to santize the feature columns first.  
  7. Next, we use Spark CSV parser to load the csv to Data Frames,
  1.     //load train data and test data from CSV Files
  2.     val trainData: DataFrame = loadCSVFile("src/test/resources/data/adult.data.csv")
  3.     val testData: DataFrame = loadCSVFile("src/test/resources/data/adult.test.txt")
  5.     //set the feature column names to the Data Frames.
  6.     val train = trainData.toDF(features: _*)
  7.     val test = testData.toDF(features: _*)
  1. Once we load the data frames, we can put them into NamedDataFrames, then we are ready for next steps. 
  1.     //create named dataframes
  2.     val dataframes = List(NamedDataFrame(name = "train", train), NamedDataFrame(name = "test", test))

  • part 2: generate feature statistics and save to files

    Here are saved two version of files, one is binary, andother one is base64 encoded binary file.
    If one uses the later version, you won’t need to encoded into base64 in Juypter notebook. 

  1.     val proto = generator.protoFromDataFrames(dataframes)
  3.     //persist protobuf binary into files (without or with base64 encoding)
  4.     persistProto(proto, base64Encode = false, new File("src/test/resources/data/stats.pb"))
  5.     persistProto(proto, base64Encode = true, new File("src/test/resources/data/stats.txt"))

The complete code can be found in
Overview Demo

  • Part 3 : load protobuf to Jupyter Notebook

    Now, look at the Jupyter Notebook

    We load the “stats.pb.txt” file to a string. Since the stats.pb.txt is simply base64encoded protobuf stats.pb
    we just directly load it to string and pass to HTML Template.

As noted in Facets

For using Facets Overview in a Jupyter notebook, you will need to change the path that the Facets Overview python
code is loaded from to the correct path given where your notebook kernel is run from.

Assuming that you are going to start the notebook kernel on project directory,

In the code, we specify the path as “../data/stats.pb.txt” as the notebook file *.ipynb
is located at /demo/python/ directory; and test file “stats.pb.txt” is in /demo/data/ directory

If you want to move the files in locations or start notebook kernel in different locations, you need to adjust those paths.

When visualizing a large amount of data in a Juypter notebook,you will need to start the notebook server with
an increased IOPub data rate. This can be done with the command 

  1.    jupyter notebook --NotebookApp.iopub_data_rate_limit=10000000

How to visualize the generated feature statistics in browser (Javascripts)

Google Facets (https://github.com/PAIR-code/facets) has provided Javascripts visualization,
it is located in different branch (rather than master branch).

This can be seen in demo page works (https://pair-code.github.io/facets/)
and the code for that page here: https://github.com/PAIR-code/facets/blob/gh-pages/index.html#L231

The difference here is that we don’t re-generate the features stats from existing file, it will load the
probuff string from the file we already generated.

Facets has made this simple as you can seen the index.html, here we use JQuery to load the protobuf file
into string and assign to HTML element. You can simply use other way to do it.

  1.     <!DOCTYPE html>
  2.     <html lang="en-US">
  3.     <head>
  4.         <meta http-equiv="Content-type" content="text/html;charset=UTF-8"/>
  5.         <title>Visualizations for ML datasets</title>
  7.         <!-- JS includes -->
  8.         <script src="https://ajax.googleapis.com/ajax/libs/jquery/3.2.1/jquery.min.js"></script>
  9.         <script src="https://cdnjs.cloudflare.com/ajax/libs/webcomponentsjs/1.3.3/webcomponents-lite.js"></script>
  10.         <link rel="import" href="https://raw.githubusercontent.com/PAIR-code/facets/1.0.0/facets-dist/facets-jupyter.html" >
  12.     </head>
  14.     <body>
  15.     <facets-overview id="elem"></facets-overview>
  17.     <script>
  19.       function setupVis() {
  20.         $.when(protoAjax()).done(function() {
  21.            console.log("ok");
  22.         });
  24.         function protoAjax() {
  25.          var jqxhr = $.get( {url: "../data/stats.pb.txt"})
  26.             .done(function(data) {
  27.                 var overview = document.querySelector("#elem");
  28.                 overview.protoInput = data;
  29.             })
  30.         }
  31.       }
  33.       $( document ).ready(function() {
  34.         setupVis();
  35.       });
  36.     </script>
  37.     </body>
  39.     </html>

If you want to try this out, you can start a http server

Using a HTTP server for python, you can do the following 

  1.     python -m http.server

Then, in the browser put the URL to

http://0.0.0.0:8000/demo/javascripts/index.html

if you like to load different datasets, you can generate a new protobuf and change
above path. Or add javascripts function to change the paths.

Development

This is maven project, so it usually follows the standard maven commands for build.

Build

  1.     mvn clean package
  1. * this works for Apache Spark 2.4.x (branch v0.4.1), but we need additional steps for Apache Spark 3
  2. * see section on Apache Spark 3

Test

  1.     mvn test

License

  1.     Apache License 2.0

API Usage Notes.

pay attention to categorical features

API FeatureStatsGenerator.protoFromDataFrames() takes three arguments

  • dataFrames — List of NameDataFrames
  • features — white list of features
  • catHistgmLevel — categorical feature histogram level 
  3.   def protoFromDataFrames(dataFrames     : List[NamedDataFrame],
  4.                           features       : Set[String] = Set.empty[String],
  5.                           catHistgmLevel : Option[Int] = Some(20): DatasetFeatureStatisticsList = {

suggest pay extra attention to catHistgmLevel.
If the feature is categorical feature such as user email or with high number of unique values, then set catHistgmLevel to None
will cause the resulting protobuf file size to include all unique values of the categorical feature,
depending on the data size, this can be very large. For data size with 1-2 millions rows, the result can be several
hundreds MB if you have more than one such features. The UI mearly use these to show raw data, so set
catHistgmLevel = Some(20) should be enough. This can significant reduce the result file size.

pay attention to data types

  • The library takes the data types of the feature from DataFrame schema. If Spark derives the schema from the data using Spark SQL,
    the schema may not accurate, the numerical data or integer data may become as other types depending on the data,
  1.  val df = spark.sql("select * from mytable")

in many cases, it is necessary to explicit set the data type in SQL, such as

  1.  select 
  2.       field1 
  3.     , cast (field2 as double) as field2
  4.     , cast (field3 as Int) as field3
  5.     from mytable

  • boolean data type needs to be converted to integer type 

    1.   select case when booleanfield = true then 1 else 0 end as booleanfield
    2.   from mytable
    • Date/Datetime data type
      The facets-overview doesn’t handle Date/Datatime data type, therefore,
      one needs to cast date/datatime to categorical data type before pass the data to facets-overview

Apache Spark 3 support and branches

note we have upgraded the code to support Apache spark 3. To do so, we created the following branches

  • v0.4.1 branch is used for Apache Spark 2.4.x, Scala 2.11

  • main branch is now used for Apache Spark 3, and Scala 2.12

    Apache Spark 3 requires scala 2.12. Facets-overview-spark depends on Spark-tensorflow-connector.
    At this time (2021-06-02), spark-tensorflow-connector has no release jar available at maven central repository for Scala 2.12
    so we temporarily add git submodule to the project, so we can build Spark-tensorflow-connector locally

    The details of git submodule can be found at submodule,
    when clone the project, you need to 

    1.   git clone --recurse-submodules git@github.com:gopro/facets-overview-spark.git

    to include tensorflow ecosystem submodule, then

    1.    cd ecosystem/spark/spark-tensorflow-connector;
    2.    mvn clean install

    once you have the jar created in local mvn repository, then you can build normally in the current project.

    1.     cd facets-overview-spark
    2.     mvn clean package

    If you already have cloned the project without —recurse-submodules.
    You need to do git submodule init and git submodule update as described in above documentation