DCausse has submitted this change and it was merged. (
https://gerrit.wikimedia.org/r/364638 )
Change subject: Add some feature engineering helper functions
......................................................................
Add some feature engineering helper functions
A couple small functions to assist with feature engineering,
and some documentation of how to create and test new features.
Change-Id: Ic87b0a48c94c5b8f449cebb496626cdb2cb90011
---
A docs/feature_engineering.rst
M mjolnir/feature_engineering.py
2 files changed, 218 insertions(+), 2 deletions(-)
Approvals:
DCausse: Verified; Looks good to me, approved
diff --git a/docs/feature_engineering.rst b/docs/feature_engineering.rst
new file mode 100644
index 0000000..41416a0
--- /dev/null
+++ b/docs/feature_engineering.rst
@@ -0,0 +1,158 @@
+This document contains examples of testing new features with mjolnir to
+hopefully guide future experimentation with new features quickly, without
+having to import all the relevant data or create new fields in elasticsearch
+for the initial evaluation.
+
+Features that seem to be useful should then be implemented as part of the
+elasticsearch feature collection pipeline and re-verified.
+
+
+Evaluating new query independent features
+=========================================
+
+The example calculates a new query independent feature in mysql and appends it
+to an existing test and train dataset.
+
+Calculate a potential new feature from mysql. It is important to check a mysql
`explain` before
+running your feature collction query to ensure it is not overly expensive.
Visiting every row
+of the enwiki revisions table for example would take perhaps hours::
+
+ echo 'select rev_page, min(rev_timestamp) from enwiki.revision group by
rev_page' | \
+ mysql -sN
--defaults-extra-file=/etc/mysql/conf.d/analytics-research-client.cnf -h
analytics-store > page_min_timestamp
+
+Open a pyspark shell with mjolnir::
+
+ PYSPARK_PYTHON=venv/bin/python SPARK_CONF_DIR=/etc/spark/conf
~/spark-2.1.0-bin-hadoop2.6/bin/pyspark \
+ --jars /home/ebernhardson/mjolnir-0.1-jar-with-dependencies.jar \
+ --driver-class-path
/home/ebernhardson/mjolnir-0.1-jar-with-dependencies.jar \
+ --master yarn \
+ --files /usr/lib/libhdfs.so.0.0.0 \
+ --archives 'mjolnir_venv.zip#venv'
+
+Load the feature file into a dataframe. Specifying the schema allows naming
the columns
+and ensuring they are converted to the correc types::
+
+ import pyspark.sql.types
+ df_mwts = sqlContext.read.csv(
+ '/path/to/page_min_timestamp',
+ sep='\t', schema=pyspark.sql.types.StructType([
+ pyspark.sql.types.StructField('hit_page_id',
pyspark.sql.types.IntegerType()),
+ pyspark.sql.types.StructField('mwts',
pyspark.sql.types.StringType()),
+ ]))
+
+Munge the feature from a mediawiki timestamp into a normal unix timestamp::
+
+ import datetime
+ from pyspark.sql import functions as F
+ import time
+ mwts_to_ts_udf = F.udf(lambda mwts:
time.mktime(datetime.datetime.strptime(mwts, '%Y%m%d%H%M%S').timetuple()),
pyspark.sql.types.FloatType())
+ df_page_created = df_mwts.select('hit_page_id',
mwts_to_ts_udf('mwts').alias('page_created'))
+
+Load existing df generated by data_pipeline.py and merge new feature::
+
+ df_train =
sqlContext.read.parquet('hdfs://analytics-hadoop/user/ebernhardson/training_size/train_1193k')
+ df_test =
sqlContext.read.parquet('hdfs://analytics-hadoop/user/ebernhardson/training_size/test')
+
+Append the new feature to existing features::
+
+ import mjolnir.feature_engineering
+ def merge(df):
+ df_joined = df.join(df_page_created, how='inner', on=['hit_page_id'])
+ return mjolnir.feature_engineering.append_features(df_joined,
'page_created').drop('page_created')
+
+
merge(df_train).write.parquet('hdfs://analytics-hadoop/user/ebernhardson/mjolnir/1193k_with_page_created')
+
merge(df_test).write.parquet('hdfs://analytics-hadoop/user/ebernhardson/mjolnir/test_with_page_created')
+
+Now we have two new datasets that contain the new feature, one for training
and one for evaluation. The training set can be fed into training_pipeline.py
as the --input parameter. The test set can be provided as --test-dir to be
evaluated against the model after training is complete.
+
+Calculating a new query dependent feature from the elasticsearch analyze api
+============================================================================
+
+Open a pyspark shell with mjolnir::
+
+ PYSPARK_PYTHON=venv/bin/python SPARK_CONF_DIR=/etc/spark/conf
~/spark-2.1.0-bin-hadoop2.6/bin/pyspark \
+ --jars /home/ebernhardson/mjolnir-0.1-jar-with-dependencies.jar \
+ --driver-class-path
/home/ebernhardson/mjolnir-0.1-jar-with-dependencies.jar \
+ --master yarn \
+ --files /usr/lib/libhdfs.so.0.0.0 \
+ --archives 'mjolnir_venv.zip#venv'
+
+Load all the relevant query strings into an rdd::
+
+ df_1193k =
sqlContext.read.parquet('hdfs://analytics-hadoop/user/ebernhardson/mjolnir/training_size/train_1193k')
+ df_test =
sqlContext.read.parquet('hdfs://analytics-hadoop/user/ebernhardson/mjolnir/training_size/test')
+ rdd_queries = sc.union([df.select('query').rdd for df in [df_1193k,
df_test]).distinct()
+
+Define a function that will iterate over rows in a partition and collect our
+new feature. Because the query issued is increadibly light weight and the
+execution time is dominated by round trips between datacenters it utilizes a
+thread pool to further parallelize the work on each partition. Note that when
+using pyspark in a shell it has issues automagically importing from multiple
+levels, so the `import multiprocessing.dummy` has to go inside the function run
+on each worker. Additionally a `requests.Session` is used to ensure http
+keep-alive is used, and one is created per-thread as it is not thread safe.::
+
+ import json
+ import requests
+ import random
+ import threading
+ def gen_collect_partition(index, analyzer)
+ def f(rows):
+ import multiprocessing.dummy
+ url = 'http://elastic%d.codfw.wmnet:9200/%s/_analyze' %
(random.randint(2001, 2035), index)
+ tl = threading.local()
+ def init():
+ tl.session = requests.Session()
+ def process_one(row):
+ res = tl.session.get(url, data=json.dumps({
+ 'analyzer': analyzer,
+ 'text': row.query,
+ })
+ return (row.query, len(res.json()['tokens']))
+ pool = multiprocessing.dummy.Pool(10, init)
+ return pool.imap_unordered(rows)
+ return f
+
+Apply that function to each partition of the rdd. This is somewhat expensive to
+compute, taking ~75s against 500k unique queries in ~30 partitions, and it will
+be used multiple times so we cache it in memory::
+
+ df_num_plain_tokens =
rdd.mapPartitions(gen_collect_partition('enwiki_content',
'text_search')).toDF(['query', 'num_text_tokens'])
+ df_num_text_tokens =
rdd.mapPartitions(gen_collect_partition('enwiki_content',
'plain_search')).toDF(['query', 'num_plain_tokens'])
+ df_num_tokens = df_num_plain_tokens.join(df_num_text_tokens, how='inner',
on=['query']).cache()
+
+Merge our new feature into the existing data sets and write them out to hdfs::
+
+ import mjolnir.feature_engineering
+ def merge(df):
+ df_joined = df.join(df_num_tokens, how='inner', on=['query'])
+ return mjolnir.feature_engineering.append_features(df_joined,
'num_text_tokens',
'num_plain_tokens').drop('num_query_tokens').drop('num_plain_tokens')
+
+
merge(df_1193k).write.parquet('hdfs://analytics-hadoop/user/ebernhardson/mjolnir/1193k_with_num_query_tokens')
+
merge(df_test).write.parquet('hdfs://analytics-hadoop/user/ebernhardson/mjolnir/test_with_num_query_tokens')
+
+Again we have two new datasets with an additional feature that can be evaluated
+with training_pipeline.py. These new features can be tested together directly::
+
+ PYSPARK_PYTHON=venv/bin/python SPARK_CONF_DIR=/etc/spark/conf
~/spark-2.1.0-bin-hadoop2.6/bin/spark-submit \
+ --jars /home/ebernhardson/mjolnir-0.1-jar-with-dependencies.jar \
+ --driver-class-path
/home/ebernhardson/mjolnir-0.1-jar-with-dependencies.jar \
+ --master yarn \
+ --files /usr/lib/libhdfs.so.0.0.0 \
+ --conf spark.dynamicAllocation.maxExecutors=105 \
+ --conf spark.sql.autoBroadcastJoinThreshold=-1 \
+ --conf spark.task.cpus=4 \
+ --conf spark.yarn.executor.memoryOverhead=1536 \
+ --executor-memory 2G i\
+ --executor-cores 4 \
+ --archives 'mjolnir_venv.zip#venv' \
+ venv/lib/python2.7/site-packages/mjolnir/cli/training_pipeline.py \
+ -i
hdfs://analytics-hadoop/user/ebernhardson/mjolnir/1193k_with_num_query_tokens \
+ -o /home/ebernhardson/training_size/1193k_with_num_query_tokens \
+ -t
hdfs://analytics-hadoop/user/ebernhardson/mjolnir/test_with_num_query_tokens
+ -w 1 -c 100 -f 5 enwiki
+
+The new features can also be tested individually by using `--zero-feature`
+argument to training_pipeline.py to zero out the feature not being evaluated.
+
+
diff --git a/mjolnir/feature_engineering.py b/mjolnir/feature_engineering.py
index 8b942d2..8301547 100644
--- a/mjolnir/feature_engineering.py
+++ b/mjolnir/feature_engineering.py
@@ -1,12 +1,70 @@
+"""Helpful utilities for feature engineering"""
import numpy as np
+import mjolnir.spark
from pyspark.ml.linalg import Vectors, VectorUDT
+from pyspark.sql import functions as F
+import pyspark.sql.types
+
def append_features(df, *cols):
+ """Append features from columns to the features vector.
+
+ Parameters
+ ----------
+ df : pyspark.sql.DataFrame
+ cols : list of str
+
+ Returns
+ -------
+ pyspark.sql.DataFrame
+ """
def add_features(feat, *other):
raw = feat.toArray()
return Vectors.dense(np.append(raw, map(float, other)))
add_features_udf = F.udf(add_features, VectorUDT())
new_feat_list = df.schema['features'].metadata['features'] + cols
- return df.withColumn('features', mjolnir.spark.add_meta(df._sc,
add_features_udf('features', *cols),
- {'features':
new_feat_list}))
+ return df.withColumn('features', mjolnir.spark.add_meta(
+ df._sc, add_features_udf('features', *cols), {'features':
new_feat_list}))
+
+def zero_features(df, *feature_names):
+ """Zero out features in the feature vector.
+
+ Parameters
+ ----------
+ df : pyspark.sql.DataFrame
+ feature_names : list of str
+
+ Returns
+ -------
+ pyspark.sql.DataFrame
+ """
+ features = df.schema['features'].metadata['features']
+ idxs = [features.index(name) for name in feature_names]
+ def zero_features(feat):
+ raw = feat.toArray()
+ for idx in idxs:
+ raw[idx] = 0.
+ return Vectors.dense(raw)
+ zero_features_udf = F.udf(zero_features, VectorUDT())
+ return df.withColumn('features', mjolnir.spark.add_meta(
+ df._sc, zero_features_udf('features'), {'features': features}))
+
+
+def explode_features(df):
+ """Convert feature vector into individual columns
+
+ Parameters
+ ----------
+ df : pyspark.sql.DataFrame
+
+ Returns
+ -------
+ pyspark.sql.DataFrame
+ """
+ features = df.schema['features'].metadata['features']
+ def extract_feature(features, idx):
+ return features.toArray()[idx]
+ extract_feature_udf = F.udf(extract_feature, pyspark.sql.types.FloatType())
+ cols = [extract_feature_udf('features', F.lit(idx)).alias(name) for idx,
name in enumerate(features)]
+ return df.select('*', *cols)
--
To view, visit https://gerrit.wikimedia.org/r/364638
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: merged
Gerrit-Change-Id: Ic87b0a48c94c5b8f449cebb496626cdb2cb90011
Gerrit-PatchSet: 2
Gerrit-Project: search/MjoLniR
Gerrit-Branch: master
Gerrit-Owner: EBernhardson <[email protected]>
Gerrit-Reviewer: DCausse <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
