Update app.py

app.py

@@ -9,6 +9,7 @@ import pyarrow as pa
 import pyarrow.parquet as pq
 import math
 import re
+from concurrent.futures import ThreadPoolExecutor, as_completed
 # Set page config for a wider layout and custom theme
 st.set_page_config(layout="wide", page_title="Job Listings Dashboard")
 
@@ -53,56 +54,51 @@ HF_TOKEN = st.secrets["HF_TOKEN"]
 HF_USERNAME = st.secrets["HF_USERNAME"]
 DATASET_NAME = "jobeasz"
 
+
 @st.cache_data(ttl=3600)
 def load_and_concat_data():
     api = HfApi()
     dataset_files = api.list_repo_files(repo_id=f"{HF_USERNAME}/{DATASET_NAME}", repo_type="dataset")
     csv_files = [file for file in dataset_files if file.endswith('.csv')]
 
-    all_data = []
-    for file in csv_files:
+    def process_file(file):
         try:
             file_content = api.hf_hub_download(repo_id=f"{HF_USERNAME}/{DATASET_NAME}", filename=file, repo_type="dataset", token=HF_TOKEN)
-            df = pd.read_csv(file_content, engine='pyarrow')
-            all_data.append(df)
-        except Exception:
-            pass  # Silently skip files that can't be processed
+            df = pd.read_csv(file_content, engine='pyarrow', usecols=[
+                'site', 'job_url', 'title', 'company', 'location',
+                'job_type', 'date_posted', 'is_remote', 'company_url'
+            ])
+            df['date_posted'] = pd.to_datetime(df['date_posted'], errors='coerce')
+            df = df[df['date_posted'].dt.year == 2024].dropna(subset=['date_posted'])
+            df['title'] = df['title'].str.lower()
+            df['company'] = df['company'].str.lower()
+            df['location'] = df['location'].apply(clean_location)
+            return df
+        except Exception as e:
+            print(f"Error processing file {file}: {str(e)}")
+            return None
+
+    def clean_location(location):
+        if pd.isna(location):
+            return location
+        location = location.lower()
+        return re.sub(r',\s*(us|usa)$', '', location)
+
+    # Use ThreadPoolExecutor for parallel processing
+    with ThreadPoolExecutor(max_workers=4) as executor:
+        future_to_file = {executor.submit(process_file, file): file for file in csv_files}
+        all_data = []
+        for future in as_completed(future_to_file):
+            df = future.result()
+            if df is not None:
+                all_data.append(df)
 
     if not all_data:
         return pd.DataFrame()
 
     concatenated_df = pd.concat(all_data, ignore_index=True)
-    
-    columns_to_keep = [
-        'site', 'job_url', 'title', 'company', 'location',
-        'job_type', 'date_posted', 'is_remote', 'company_url'
-    ]
-    filtered_df = concatenated_df[columns_to_keep].reset_index(drop=True)
-    filtered_df['date_posted'] = pd.to_datetime(filtered_df['date_posted'], errors='coerce')
-    
-    # Drop duplicates and rows with NaT in date_posted
-    filtered_df = filtered_df.drop_duplicates().dropna(subset=['date_posted'])
-    #filtering based on data in 2024
-    filtered_df = filtered_df[filtered_df['date_posted'].dt.year==2024]
-    # Convert titles and company name to lowercase
-    filtered_df['title'] = filtered_df['title'].str.lower()
-    filtered_df['company'] = filtered_df['company'].str.lower()
-    
-    # Function to clean the location
-    def clean_location(location):
-        if pd.isna(location):
-            return location  # Return NaN as is
-        # Convert to lowercase
-        location = location.lower()
-        # Remove ', us' or ', usa' from the end using regex
-        location = re.sub(r',\s*(us|usa)$', '', location)
-        return location
-
-    # Clean the location in place
-    filtered_df['location'] = filtered_df['location'].apply(clean_location)
-    #added new line to drop duplciate records
-    filtered_df = filtered_df.drop_duplicates()
-    
+    filtered_df = concatenated_df.drop_duplicates().reset_index(drop=True)
+
     return filtered_df
     
 @st.cache_data()