保留正文与补采附件。源页面当前未重新核验；具体限制见交付清单。

Revisit consent button We value your privacy We use cookies to enhance your browsing experience, serve personalised ads or content, and analyse our traffic. By clicking "Accept All", you consent to our use of cookies. Customise Reject All Accept All Customise Consent Preferences We use cookies to help you navigate efficiently and perform certain functions. You will find detailed information about all cookies under each consent category below. The cookies that are categorised as "Necessary" are stored on your browser as they are essential for enabling the basic functionalities of the site. ... Show more NecessaryAlways Active Necessary cookies are required to enable the basic features of this site, such as providing secure log-in or adjusting your consent preferences. These cookies do not store any personally identifiable data.

Functional Functional cookies help perform certain functionalities like sharing the content of the website on social media platforms, collecting feedback, and other third-party features.

Analytics Analytical cookies are used to understand how visitors interact with the website. These cookies help provide information on metrics such as the number of visitors, bounce rate, traffic source, etc.

Performance Performance cookies are used to understand and analyse the key performance indexes of the website which helps in delivering a better user experience for the visitors.

Advertisement Advertisement cookies are used to provide visitors with customised advertisements based on the pages you visited previously and to analyse the effectiveness of the ad campaigns.

Uncategorised Other uncategorised cookies are those that are being analysed and have not been classified into a category as yet. No cookies to display. Reject All Save My Preferences Accept All Toggle SearchToggle Mobile Navigation Publish AI, ML & data-science insights to a global community of data professionals. Sign inSubmit an Article LatestEditor's PicksNewsletterShipAI

Write For TDS LinkedInXToggle Search Data Science

Feature Engineering for Time-Series Using PySpark on Databricks

Discover the potentials of PySpark for time-series data: Ingest, extract, and visualize data, accompanied by practical implementation codes John Leung May 15, 20249 min read

Discover the potentials of PySpark for time-series data: Ingest, extract, and visualize data, accompanied by practical implementation code

With the increasing demand for high-speed querying and analysis on large datasets, Apache Spark has stood out as one of the most popular analytical engines in recent years. It is powerful in distributed data processing due to its master-worker architecture. This includes a driver program that coordinates with the cluster manager (master) and controls the execution of distributing smaller tasks to worker nodes. Besides, designed as an in-memory data processing engine, Spark primarily uses RAM to store and manipulate data, rather than relying on disk storage. These coordinately facilitate faster execution of overall tasks. Photo by Dawid Zawiła on Unsplash Photo by Dawid Zawiła on Unsplash Learn this step by step with the interactive AI and Data Scientist roadmap.

Apache Spark: From low-level to high-level

At the lower level, its architecture is designed based on two main abstractions:

At the higher level, we can leverage a rich set of high-level tools using languages Scala, Python, or R. Examples of tools include Spark SQL for SQL and DataFrames, Pandas API on Spark for Pandas workloads, and Structured Streaming for stream processing. However, before enjoying these functionalities, we may need much effort to self-manage a Spark cluster with the setup of infrastructure and a bunch of complex tools, which could cause a headache.

PySpark on Databricks

To address these challenges, PySpark on Databricks is recently one of the high-level solutions in the industry. PySpark is the Python API for Spark, while Databricks is a full software platform built on top of Spark. It includes notebooks, infrastructure orchestration (auto-provisioning and scaling), process orchestration (job submission and scheduling), managed clusters, and even source control.

Using PySpark APIs in Databricks, we will demonstrate and perform a feature engineering project on time series data. In this hands-on journey, we will simulate how Pandas library generally behaves for data processing, with the extra benefits of scalability and parallelism. Note: If you want to know further how to dynamically orchestrate this Databricks notebook written in PySpark APIs in Azure, you can click here . Photo by Alexandru Boicu on Unsplash Photo by Alexandru Boicu on Unsplash Consider a scenario in which you have the electric power consumption data for your household on hand, sampled at a one-minute rate from December 2006 to November 2010. Our objective is to ingest and manipulate data, extract features, and generate visualizations. The dataset [with license as Database: Open Database, Contents: Database Contents], obtained from Kaggle, includes various fields such as date, time, global power (active and reactive), voltage, global intensity, and submetering (1, 2 and 3). We can begin our analysis now.

The initial setup

To begin, we need to create a user account for Databricks Community Edition, which gives a suitable Databricks environment for our proof-of-concept purpose. Afterward, we can upload the input data file to the FileStore, a dedicated Databricks path. By clicking "Create Table in Notebook", you are provided with the code template to initiate data ingestion. Initial setup 1/2 - Create a user account Image by author Initial setup (1/2) - Create a user account (Image by author) Initial setup 2/2 - Create a new table Image by author Initial setup (2/2) - Create a new table (Image by author)

Create a feature engineering project

#1 Ingest data

We use the method spark.read() to read our data source and return a DataFrame, a relational table. It supports various data sources such as CSV, JSON, Parquet, and more. In this case, we read the power consumption data in CSV format with a defined schema, where the first row serves as the header, and ";" is as the delimiter. python

# File location and type
file_location = "/FileStore/tables/household_power_consumption.csv"
file_type = "csv"

# CSV options
schema = "Date STRING, Time STRING, Global_active_power DOUBLE, Global_reactive_power DOUBLE, Voltage DOUBLE, Global_intensity DOUBLE, Sub_metering_1 DOUBLE, Sub_metering_2 DOUBLE, Sub_metering_3 DOUBLE"
first_row_as_header = "true"
delimiter = ";"

# Read CSV files
org_df = spark.read.format(file_type) 
.schema(schema) 
.option("header", first_row_as_header) 
.option("delimiter", delimiter) 
.load(file_location)

display(org_df)

The output of the DataFrame with the first several rows: Output of the DataFrame Image by author Output of the DataFrame (Image by author)

In scenarios where data is continuously generated, we use stream processing techniques to read it incrementally. To demonstrate Spark's behavior, I partitioned the original dataset into 10 subsets and stored them at the path "/FileStore/tables/stream/" beforehand. We then use another method spark.readStream()for streaming data. python

sourceStream=spark.readStream.format("csv") 
.option("header",True) 
.schema(schema) 
.option("mode","dropMalformed") 
.option("maxFilesPerTrigger",1) 
.option("ignoreLeadingWhiteSpace",True) 
.load("dbfs:/FileStore/tables/stream")

It is worth mentioning that the mode setting as "dropMalformed" means that we discard the corrupted records, no matter whether the corruption is due to structural inconsistencies or other factors that make them unusable. Also, we choose to process only one file per trigger event. By starting to receive data and checking the record count every ten seconds, we can observe the continuous arrival of streaming data. python

import time

# Stream the content of the DataFrame
query = sourceStream.writeStream 
.queryName("count") 
.format("memory") 
.outputMode("append") 
.start()

# Display the count of rows
for _ in range(10):
  spark.sql("SELECT COUNT(*) AS no_of_rows FROM count").show()
  time.sleep(10)

#2 Manipulate and explore data

Since the number of rows with missing values is relatively negligible, we choose to drop them. Besides, we extract time-related features so that patterns can potentially be observed in higher dimensions later. python

from pyspark.sql.functions import col, concat_ws, to_date

# Drop rows with missing values
df = org_df.na.drop()

# Convert columns "Date" and "Time" into new column "DateTime"
df = df.withColumn("Date", to_date(col("Date"),"d/M/y"))
df = df.withColumn("Date", df["Date"].cast("date"))
df = df.select(concat_ws(" ", to_date(col("Date"),"d/M/y"), col("Time")).alias("DateTime"), "*")
df = df.withColumn("DateTime", df["DateTime"].cast("timestamp"))

# Add time-related features
df = df.withColumn("year", year("DateTime"))
df = df.withColumn("month", month("DateTime"))
df = df.withColumn("week_num", weekofyear("DateTime"))
df = df.withColumn("hour", hour("DateTime"))

We can explore data with various basic PySpark methods. (1) Select The "'select" method allows us to create a subset of the data frame column-wise. In this example, we select columns in descending order of global active power. python

df.select("DateTime", "Global_active_power", "Global_intensity").sort("Global_active_power", ascending=False).show(5)

Output of select method Image by author Output of "select" method (Image by author) (2) Filter This filters data points based on column values. In this example, we filter by two columns: "year" and "Global_intensity". python

df.filter(
    (col("year") == 2009) &
    (col("Global_intensity") > 40)
).count()

# Output: 10

(3) groupby We can also perform some aggregations. In our dataset, we calculate the average of global active power and sub-meterings for different months. python

df.groupby("month").agg(
     round(mean("Global_active_power"), 2).alias("Avg_global_active_power"),
     round(mean("Sub_metering_1"), 2).alias("Avg_sub_metering_1"),
     round(mean("Sub_metering_2"), 2).alias("Avg_sub_metering_2"),
     round(mean("Sub_metering_3"), 2).alias("Avg_sub_metering_3"),
).sort(["month"]).show(5)

Output of groupby method Image by author Output of "groupby" method (Image by author) #3 Extract features using Window functions In addition to the above basic PySpark methods and functions, we can leverage Window functions to generate additional features to capture temporal dependencies and relationships in time series data. Assuming we have a transformed dataset ("df2") where the total global active power is aggregated by day from one-minute rate samples. Let's explore how we can obtain these features. (1) Lag features These represent the metrics' values from previous days, which aids our models in learning from historical data and identifying trends. python

from pyspark.sql.window import Window
from pyspark.sql.functions import lag, round

# Create a Window specification based on the 'Date' column
windowSpec = Window.orderBy("Date")

# Calculate the lagged value of 'Total_global_active_power'
df2 = df2.withColumn("power_lag1", round(lag(col("Total_global_active_power"), 1).over(windowSpec), 2))

display(df2)

Output - Lag features Image by author Output - Lag features (Image by author) (2) Delta features This is to take a subsequent step to capture short-term changes or variations over time by calculating the difference between original data fields and the lag features. python

# Calculate the difference between columns
df2 = df2.withColumn("power_lag1_delta", round(col("power_lag1") - col("Total_global_active_power"), 2))

display(df2)

Output - Delta features Image by author Output - Delta features (Image by author) (3) Window average features These features calculate the average value of our target data fields within a sliding window, enabling us to capture the smoothed patterns and relatively long-term trends. Here I pick the window sizes as 14 (2 weeks) and 30 (roughly 1 month). python

# Add window average fields to the DataFrame for the specified window sizes
def add_window_avg_features(df, window_sizes):
    for window_size in window_sizes:
        window_col_name = f"avg_power_l{window_size}"
        windowSpec = Window.orderBy("Date").rowsBetween(-window_size, 0)
        df = df.withColumn(window_col_name, round(avg(col("Total_global_active_power")).over(windowSpec), 2))
    return df

window_sizes = [14, 30]
df2 = add_window_avg_features(df2, window_sizes)

df2.select("Date", "Total_global_active_power", "avg_power_l14", "avg_power_l30").sort("Date", ascending=False).show(5)

Output - Window average features Image by author Output - Window average features (Image by author) (4) Exponentially weighted moving average (EWMA) features EWMA features are the rectified version of window average features by assigning more weight and emphasis to recent data, and less to past data. A higher value of weight (alpha) means that the EWMA features track more closely to the original time series. Here I pick two separate weights: 0.2 and 0.8. python

import pyspark.pandas as ps

#  Add EWMA features to the DataFrame for the specified alpha values
def add_ewma_features(df, alphas):
    for alpha in alphas:
        ewma_col_name = f"ewma_power_w{str(alpha).replace('.', '')}"
        windowSpec = Window.orderBy("Date")
        df[ewma_col_name] = df.Total_global_active_power.ewm(alpha=alpha).mean().round(2)
    return df

alphas = [0.2, 0.8]
# Convert into a pandas-on-Spark DataFrame, to use EWM function
df2_pd = df2.pandas_api()
df2_pd = add_ewma_features(df2_pd, alphas)
# Convert back to a Spark DataFrame
df2 = df2_pd.to_spark()

df2.select("Date", "Total_global_active_power", "ewma_power_w02", "ewma_power_w08").sort("Date", ascending=False).show(5)

Output - EWMA features Image by author Output - EWMA features (Image by author) #4 Generate visualizations on Notebook After extracting time-related data and features using various PySpark functions and methods, we can use the built-in support from Databricks to create visualizations efficiently. This works by dragging and dropping data fields and configuring visual settings in the visualization editor. Some examples are shown below.

Interpretation: There is a highly positive correlation between the two fields. Scatter plot, using a visualization editor Image by author Scatter plot, using a visualization editor (Image by author)

Interpretation: There are relatively large variations of global active power from 7:00 to 21:00. Box plot Image by author Box plot (Image by author)

Interpretation: EWMA with alpha 0.8 sticks more closely to the original time series than EWMA with alpha 0.2. Line chart Image by author Line chart (Image by author) Besides, we can generate the default data profiles to display summary statistics such as count, % of missing values, and data distributions. This ensures data quality throughout the entire feature engineering process. These above visualizations can alternatively be created by the query output using Databricks SQL.

Wrapping it up

In our hands-on exploration, we used PySpark for feature engineering of time-series data using the Databricks platform:

When dealing with a massive dataset, PySpark is often preferred over Pandas due to its scalability and performance capabilities. PySpark's support for lazy evaluation also means that computations are only performed when necessary, which reduces the overhead. However, Scala can sometimes be a better option, because we can closely catch up with the latest features as Spark itself is written in Scala. And more likely that systems which are less error-prone can be designed with the use of immutable objects. Different languages or libraries have their edges. Ultimately, the choice depends on the enterprise requirements, developers' learning curves, and integration with other systems.

Before you go

If you enjoy this reading, I invite you to **** follow my Medium page and LinkedIn page. By doing so, you can stay updated with exciting content related to data science side projects, and Machine Learning Operations (MLOps) demonstrations methodologies.

Performing Customer Analytics with LangChain and LLMs Managing the Technical Debts of Machine Learning Systems

Register Now to Continue Reading

Unlock Towards Data Science with a Free Account

All articles on Towards Data Science are available to registered users. Create a free account today to access this article and thousands more. Register for freeSign in to your account

Written By

John Leung

See all from John Leung Big DataData ScienceDatabricksPythonSpark Share This Article

Towards Data Science is a community publication. Submit your insights to reach our global audience and earn through the TDS Author Payment Program. Write for TDS

Related Articles

Your Pathway to Success: How You Can Land a Machine Learning and Data Science Internship

Machine Learning Advice and tips from a data scientist who landed two internships in a year Sara Nobrega September 4, 202415 min read

How Python Enumerations Make Data Configuration Elegant

Programming An introduction to using Enums in Python for ML projects Eirik Berge May 28, 20238 min read

Monitoring Machine Learning Models in Production: How to Track Data Quality and Integrity?

Machine Learning As the saying goes: garbage in is garbage out. Input data quality is the most crucial component of a machine learning system. Whether or... Emeli Dral January 19, 202113 min read

10 Free tools to get started with Data Visualisation-Easily & Instantly.

Data Science Jump right into the Data Visualisation process with these easy and intuitive tools. Parul Pandey March 12, 20199 min read

How to cross validate your panel data in Python

Machine Learning An introduction to panel data cross validation using PanelSplit Eric Frey March 8, 20245 min read

Data science workflows with the Targets package in R : End-to-end example with code

Data Science Structured and reproducible approach for analysis Prateek Bhatnagar August 19, 20213 min read

Why you should start using .npy file more often...

Machine Learning Native NumPy objects in Python provides speed advantage over regular objects. It works for as simple a task as reading data from file. Tirthajyoti Sarkar March 20, 20186 min read

Recreating a 538 election's plot: making custom matplotlib visualizations

Programming I have recently been interested in making highly customizable visualizations. Here is one example of trying to recreate 538's election... Dea Bardhoshi February 22, 20222 min read

The Data Scientist who rules the 'Data Science for Good' competitions on Kaggle.

Machine Learning In conversation with Shivam Bansal: A Data Scientist, a Kaggle Kernel's Grandmaster, and three times winner of Kaggle's Data Science for... Parul Pandey October 18, 20197 min read

Your home for data science and AI. The world's leading publication for data science, data analytics, data engineering, machine learning, and artificial intelligence professionals. Subscribe to Our NewsletterWrite For TDS•About•Advertise•Privacy Policy•Terms of Use © Insight Media Group, LLC 2026•The New Stack•roadmap.sh•Creator Network

Recommended Articles

Close