Sync parquet files, refresh datasets every morning

Author: brettshollenberger

config/initializers/zhong.rb

@@ -8,6 +8,12 @@
       EasyML::CleanJob.perform_later
     end
 
+    every 1.day, "refresh datasets" do
+      EasyML::Dataset.all.each do |dataset|
+        dataset.refresh_async
+      end
+    end
+
     every 1.hour, "schedule retraining" do
       EasyML::ScheduleRetrainingJob.perform_later
     end

easy_ml-0.2.0.pre.rc101.gem

@@ -16,6 +16,7 @@ class PolarsColumn
         array: Polars::List,
       }
       POLARS_MAP = {
+        Polars::Decimal => :float,
         Polars::Float64 => :float,
         Polars::Int64 => :integer,
         Polars::Float32 => :float,

lib/easy_ml/data/polars_column.rb

@@ -126,7 +126,7 @@ def download_file(object)
         )
 
         Rails.logger.info("Downloaded #{object.key} to #{local_file_path}")
-        if object.key.end_with?(".gz")
+        if object.key.end_with?(".gz") && !object.key.end_with?(".parquet.gz")
           ungzipped_file_path = ungzip_file(local_file_path)
           Rails.logger.info("Ungzipped to #{ungzipped_file_path}")
         end
@@ -284,48 +284,61 @@ def upload_file(file_path)
         relative_path = Pathname.new(file_path).relative_path_from(Pathname.new(root_dir)).to_s
         s3_key = s3_prefix.present? ? File.join(s3_prefix, File.basename(relative_path)) : relative_path
 
-        # Create a temporary gzipped version of the file
-        gzipped_file_path = "#{file_path}.gz"
-
         begin
-          Rails.logger.info("Compressing and uploading #{file_path} to s3://#{s3_bucket}/#{s3_key}")
+          Rails.logger.info("Uploading #{file_path} to s3://#{s3_bucket}/#{s3_key}")
 
-          # Compress the file
-          Zlib::GzipWriter.open(gzipped_file_path) do |gz|
+          if file_path.end_with?(".parquet")
+            # Upload parquet files directly without compression
             File.open(file_path, "rb") do |file|
-              gz.write(file.read)
+              s3.put_object(
+                bucket: s3_bucket,
+                key: s3_key,
+                body: file
+              )
+            end
+            Rails.logger.info("Successfully uploaded #{file_path} to s3://#{s3_bucket}/#{s3_key}")
+          else
+            # Create a temporary gzipped version of the file
+            gzipped_file_path = "#{file_path}.gz"
+
+            # Compress the file
+            Zlib::GzipWriter.open(gzipped_file_path) do |gz|
+              File.open(file_path, "rb") do |file|
+                gz.write(file.read)
+              end
             end
-          end
 
-          # Upload the gzipped file
-          File.open(gzipped_file_path, "rb") do |file|
-            s3.put_object(
-              bucket: s3_bucket,
-              key: "#{s3_key}.gz",
-              body: file,
-              content_encoding: "gzip",
-            )
-          end
+            # Upload the gzipped file
+            File.open(gzipped_file_path, "rb") do |file|
+              s3.put_object(
+                bucket: s3_bucket,
+                key: "#{s3_key}.gz",
+                body: file,
+                content_encoding: "gzip",
+              )
+            end
+
+            Rails.logger.info("Successfully uploaded #{file_path} to s3://#{s3_bucket}/#{s3_key}.gz")
 
-          Rails.logger.info("Successfully uploaded #{file_path} to s3://#{s3_bucket}/#{s3_key}.gz")
+            # Clean up temporary gzipped file
+            File.delete(gzipped_file_path) if File.exist?(gzipped_file_path)
+          end
         rescue Aws::S3::Errors::ServiceError, StandardError => e
           Rails.logger.error("Failed to upload #{file_path}: #{e.message}")
           raise e
-        ensure
-          # Clean up temporary gzipped file
-          File.delete(gzipped_file_path) if File.exist?(gzipped_file_path)
         end
       end
 
       def should_upload?(file_path)
         relative_path = Pathname.new(file_path).relative_path_from(Pathname.new(root_dir)).to_s
         s3_key = s3_prefix.present? ? File.join(s3_prefix, relative_path) : relative_path
+        s3_key = "#{s3_key}.gz" unless file_path.end_with?(".parquet")
 
         begin
           # Check if file exists in S3
           response = s3.head_object(
             bucket: s3_bucket,
-            key: "#{s3_key}.gz",
+            key: s3_key,
           )
 
           # Compare modification times

lib/easy_ml/data/synced_directory.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 module EasyML
-  VERSION = "0.2.0-rc100"
+  VERSION = "0.2.0-rc101"
 
   module Version
   end