I'm a newbie in PySpark, and I want to translate the Feature engineering (FE) part scripts which are pythonic, into pyspark. At first, I have Spark data frame so-called sdf including 2 columns A & B (e. g. 0 & 1 below) :
# http_path
#0 https://example.org/path/to/file?param=42#frag...
#1 https://example.org/path/to/file
# A B
#0 https://example.org/path/to/file param=42#fragment
#1 https://example.org/path/to/file NaN
Below is the example:
| data | A | B |
|---|---|---|
| https://ift.tt/3wn8yoA | path/to/file | param=42#fragment |
| https://ift.tt/3AhCxjw | path/to/file | NaN |
now I want to apply some feature engineering and extracted B1, B2, B3, B4, B5 features out of B and concat the results with sdf
#+-----------------------------------------------------------------------+
# Read file and pre-process data (parse slightly)
#+-----------------------------------------------------------------------+
import pandas as pd
import pyspark.sql.functions as f
from pyspark.sql import SQLContext
from pyspark import SparkContext, SparkConf
#sdf ----> Spark Data Frame
sdf = spark.read.csv('export.csv',inferSchema=True, header =True)
#Extract features from http_path ["api", "param"]
# regex = r'([^\?]+)\?*(.*)'
http_path = sdf.rdd.map(lambda row: row['http_path'].split('?'))
#replacing NaN or Null in case we don't have info in api or param columns
api_param_df = pd.DataFrame([[row[0], np.nan] if len(row) == 1 else row for row in http_path.collect()], columns=["api", "param"])
#Concat sdf['raw'] with api_param_df which has "api", "param" columns
pdf = pd.concat([sdf.toPandas()['raw'], api_param_df], axis=1)
# Slice columns
sdf_sliced = sdf.select("raw","api","param")
# replace empety values with Null
sdf_Null = sdf_sliced.fillna('').replace('',None)
#Drop Rows with NULL Values in Any Columns
sdf_drop = sdf_Null.na.drop(subset=["param"])
## add an index column
sdf_with_seq_id = spark.createDataFrame(sdf_drop.toPandas().reset_index()).withColumnRenamed("index","id")
#sdf_with_seq_id.show(5, truncate = False)
#pdf ----> Python Data Frame
pdf_with_seq_id = sdf_with_seq_id.toPandas().reset_index(drop=True)
pdf = pdf_with_seq_id.replace("null", np.nan).fillna(value=np.nan)
pdf
#+-----------------------------------------------------------------------+
# FE step
#+-----------------------------------------------------------------------+
#Type
def getType(input_value):
if pd.isna(input_value):
return "-"
type_ = "-"
if input_value.isdigit(): # Only numeric
type_ = "Int"
elif bool(re.match(r"^[a-zA-Z0-9_]+$", input_value)): # Consists of one or more of a-zA-Z, 0-9, underscore , and Chinese
type_ = "String"
elif bool(re.match(r"^[\d+,\s]+$", input_value)): # Only comma exists as separator "^[\d+,\s]+$"
type_ = "Array"
else:
existing_separators = re.findall(r"([\+\;\,\:\=\|\\/\#\'\"\t\r\n\s])+", input_value) # There are one or more separators
if len(existing_separators) > 1 or (len(existing_separators) == 1 and existing_separators[0] != ","): # when there is only one separator it is not comma (!= "^[\d+,\s]+$")
type_ = "Sentence"
return type_
#Length
getLength = lambda input_text: 0 if pd.isna(input_text) else len(input_text)
#Token
double_separators_regex = re.compile(r"[\<\[\(\{]+[0-9a-zA-Z_\.\-]+[\}\)\]\>]+")
single_separators_regex = re.compile(r"([0-9a-zA-Z_\.\-]+)?[\+\,\;\:\=\|\\/\#\'‘’\"“”\t\r\n\s]+([0-9a-zA-Z_\.\-]+)?")
token_number = lambda input_text: 0 if pd.isna(input_text) else len(double_separators_regex.findall(input_text) + [element for pair in single_separators_regex.findall(input_text) for element in pair if element != ""])
#Encoding type
import base64
def isBase64(input_value):
try:
return base64.b64encode(base64.b64decode(input_value)) == input_value
except Exception as e:
return False
#Character feature
N = 2
n_grams = lambda input_text: 0 if pd.isna(input_text) else len(set([input_text[character_index:character_index+N] for character_index in range(len(input_text)-N+1)]))
#frame the results of FE
features_df = pd.DataFrame()
features_df["B1"] = pdf.param.apply(getType)
features_df["B2"] = pdf.param.apply(getLength)
features_df["B3"] = pdf.param.apply(token_number)
features_df["B4"] = pdf.param.apply(isBase64)
features_df["B5"] = pdf.param.apply(n_grams)
to_numeric_columns = ["B1", "B2" , "B3", "B4", "B5"]
normalized_features_df[to_numeric_columns] = normalized_features_df[to_numeric_columns].apply(pd.to_numeric)
Until now, I could do this process by using pdf = sdf.toPandas() before FE step and convert Spark data frame to Pandas dataframe. At the end of FE step, I recreate again spark dataframe out of pdf to get benefitted from distributed processing in cluster-based platforms and use of conpet of workers for big data analysis.
from pyspark.sql import SQLContext
sc = SparkContext.getOrCreate()
sqlContext = SQLContext(sc)
columns = ["B1", "B2" , "B3", "B4", "B5"]
sdf_normalized_features_df = sqlContext.createDataFrame(normalized_features_df)
#sdf_normalized_features_df = spark.createDataFrame(normalized_features_df.astype(str), schema = columns)
sdf_normalized_features_df.show(10, truncate = False)
The expected output:
+-----------------------------------+-----------------------------------+------+------+------+------+-----+
|A |B |B1 |B2 |B3 |B4 |B5 |
+-----------------------------------+-----------------------------------+------+------+------+------+-----+
|https://example.org/path/to/file |https://example.org/path/to/file |1.0 |0.0 |0.0 |0.0 |0.0 |
|https://example.org/path/to/file |NaN |0.66 |0.06 |0.02 |0.0 |0.25 |
+-----------------------------------+-----------------------------------+------+------+------+------+-----+
Problem: what is the best approach to translating FE without converting Spark dataframe to Pandas datafarame toPandas() to optimize the pipeline and process it 100% spark form?
from How can I avoid Pandas dataframe in Spark pipeline?
No comments:
Post a Comment