hw3_final

Author: TDerig23

README.md

@@ -30,3 +30,7 @@
 # a yellow window will appear for the manual download.
 
 # use pre-commit run --all-files on ubuntu terminal to format.
+
+# use with subtable as (select \* from some table), subtable 2 as ( )
+
+# select \* from subtable

bda602_hw3.py

@@ -3,6 +3,37 @@
 from pyspark.ml.classification import LogisticRegression
 from pyspark.ml.feature import StandardScaler, VectorAssembler
 from pyspark.sql import SparkSession
+from pyspark import keyword_only
+from pyspark.ml.param.shared import HasInputCols, HasOutputCol
+from pyspark.ml.util import DefaultParamsReadable, DefaultParamsWritable
+from pyspark.sql.functions import col, concat, lit, split, when
+from pyspark.ml import Pipeline, Transformer
+
+
+class SplitColumnTransform(
+    Transformer,
+    HasInputCols,
+    HasOutputCol,
+    DefaultParamsReadable,
+    DefaultParamsWritable,
+):
+    @keyword_only
+    def __init__(self, inputCols=None, outputCol=None):
+        super(SplitColumnTransform, self).__init__()
+        kwargs = self._input_kwargs
+        self.setParams(**kwargs)
+        return
+
+    @keyword_only
+    def setParams(self, inputCols=None, outputCol=None):
+        kwargs = self._input_kwargs
+        return self._set(**kwargs)
+
+    def _transform(self, dataset):
+        input_cols = self.getInputCols()
+        output_col = self.getOutputCol()
+
+        return dataset.show()
 
 
 def main():
@@ -18,19 +49,24 @@ def main():
         .enableHiveSupport()
         .getOrCreate()
     )
-
+    # "SELECT nb.batter,nb.Hit,nb.atBat,nb.game_id,nb.local_date,"
+    # "SUM(nb.Hit) AS total_h,SUM(nb.atBat) as total_ab,(SUM(nb.Hit) / SUM(nb.atBat)) AS rolling_avg"
+    # "FROM new_baseball nb"
+    # "JOIN new_baseball nb2"
+    # "on nb.batter"
+    # "where nb.local_date between nb.local_date -100 and nb2.local_date"
+    # "GROUP by nb.batter,nb.local_date"
     sql = (
-        "SELECT nb.batter,nb.Hit,nb.atBat,nb.game_id,nb.local_date,"
-        "SUM(nb.Hit) AS total_h,SUM(nb.atBat) as total_ab,(SUM(nb.Hit) / SUM(nb.atBat)) AS rolling_avg"
-        "FROM new_baseball nb"
-        "JOIN new_baseball nb2"
-        "on nb.batter"
-        "where nb.local_date between nb.local_date -100 and nb2.local_date"
-        "GROUP by nb.batter,nb.local_date"
+        """SELECT bc.batter,bc.Hit, bc.atBat,g.game_id, g.local_dateFROM batter_counts bc,
+        SUM(nb.Hit) AS total_h,SUM(nb.atBat) as total_ab,(SUM(nb.Hit) / SUM(nb.atBat)) AS rolling_avg
+    JOIN game g
+    ON g.game_id = bc.game_id
+    order by bc.batter, bc.game_id"""
+
     )
     database = "baseball"
     user = "tderig"
-    password = ""
+    password = "password"
     server = "127.0.0.1"
     port = 3306
     jdbc_url = f"jdbc:mysql://{server}:{port}/{database}?permitMysqlScheme"
@@ -49,6 +85,21 @@ def main():
     df.show(5)
     df.printSchema()
 
+    df.createOrReplaceTempView("rolling_avg")
+    df2 = spark.sql("""select batter, game_id, SUM(Hit) AS total_h,SUM(nb.atBat) 
+                    as total_ab,(SUM(nb.Hit) / SUM(nb.atBat)) AS rolling_avg 
+                    where nb.local_date >= 2012-03-20 00:00:00.000  and nb2.local_date < 2012-06-28 22:15:00.000 
+                    GROUP by nb.batter,nb.local_date"""
+                      )
+
+
+
+    new_transform = SplitColumnTransform()
+    pipeline = Pipeline(stages=[new_transform])
+    model = pipeline.fit(df2)
+    model.transform(df2)
+
 
 if __name__ == "__main__":
     main()
+#

requirements.in

@@ -2,5 +2,7 @@ mariadb
 numpy
 pandas
 plotly
+pyspark
+pyspark-stubs
 scikit-learn
 sqlalchemy

requirements.txt

@@ -20,6 +20,14 @@ pandas==1.5.0
     # via -r requirements.in
 plotly==5.10.0
     # via -r requirements.in
+py4j==0.10.9
+    # via pyspark
+pyspark==3.0.3
+    # via
+    #   -r requirements.in
+    #   pyspark-stubs
+pyspark-stubs==3.0.0.post3
+    # via -r requirements.in
 python-dateutil==2.8.2
     # via pandas
 pytz==2022.2.1