<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Spark SQL Tables via Pyspark
Goals
Step1:  Create a SparkSession instance
Step2:  Read data from the HELK Elasticsearch via Spark SQL
Step3:  Read Sysmon Events
Step4:  Register Sysmon SQL temporary View
Step5:  Read PowerShell Events
Step6:  Register PowerShell SQL temporary View
<ASSISTANT_TASK:>
Python Code:
from pyspark.sql import SparkSession
Explanation: Spark SQL Tables via Pyspark
Goals:
Practice Spark SQL via PySpark skills
Ensure JupyterLab Server, Spark Cluster & Elasticsearch are communicating
Practice Query execution via Pyspark
Create template for future queries
Import SparkSession Class
End of explanation
spark = SparkSession.builder \
    .appName("HELK Reader") \
    .master("spark://helk-spark-master:7077") \
    .enableHiveSupport() \
    .getOrCreate()
Explanation: Create a SparkSession instance
End of explanation
es_reader = (spark.read
    .format("org.elasticsearch.spark.sql")
    .option("inferSchema", "true")
    .option("es.read.field.as.array.include", "tags")
    .option("es.nodes","helk-elasticsearch:9200")
    .option("es.net.http.auth.user","elastic")
)
    #PLEASE REMEMBER!!!!
    #If you are using elastic TRIAL license, then you need the es.net.http.auth.pass config option set
    #Example: .option("es.net.http.auth.pass","elasticpassword")
Explanation: Read data from the HELK Elasticsearch via Spark SQL
End of explanation
%%time
sysmon_df = es_reader.load("logs-endpoint-winevent-sysmon-*/")
Explanation: Read Sysmon Events
End of explanation
sysmon_df.createOrReplaceTempView("sysmon_events")
## Run SQL Queries
sysmon_ps_execution = spark.sql(
    '''
    SELECT event_id,process_parent_name,process_name
    FROM sysmon_events
    WHERE event_id = 1
        AND process_name = "powershell.exe"
        AND NOT process_parent_name = "explorer.exe"
    '''
)
sysmon_ps_execution.show(10)
sysmon_ps_module = spark.sql(
    '''
    SELECT event_id,process_name
    FROM sysmon_events
    WHERE event_id = 7 
        AND (
            lower(file_description) = "system.management.automation"
            OR lower(module_loaded) LIKE "%\\\\system.management.automation%"
        ) 
    '''
)
sysmon_ps_module.show(10)
sysmon_ps_pipe = spark.sql(
    '''
    SELECT event_id,process_name
    FROM sysmon_events
    WHERE event_id = 17
        AND lower(pipe_name) LIKE "\\\\pshost%"
    '''
)
sysmon_ps_pipe.show(10)
Explanation: Register Sysmon SQL temporary View
End of explanation
%%time
powershell_df = es_reader.load("logs-endpoint-winevent-powershell-*/")
Explanation: Read PowerShell Events
End of explanation
powershell_df.createOrReplaceTempView("powershell_events")
ps_named_pipe = spark.sql(
    '''
    SELECT event_id
    FROM powershell_events
    WHERE event_id = 53504
    '''
)
ps_named_pipe.show(10)
Explanation: Register PowerShell SQL temporary View
End of explanation
<END_TASK>