Getting it into your agent
One page per mod, every tool's command on it. A separate URL per tool would split the same page into five that compete with each other.
npx skills add jdmorag97-rgb/DDC_Skills_for_AI_Agents_in_Construction --skill parquet-convertergit clone --depth 1 https://github.com/jdmorag97-rgb/DDC_Skills_for_AI_Agents_in_ConstructionWrote this? Show the measurements
A badge with what this costs and how it scanned, read live from this page, so it follows the numbers instead of freezing them. Markdown for a README, HTML for a documentation site or a project page.
[](https://agentmods.dev/skills/jdmorag97-rgb/ddc_skills_for_ai_agents_in_construction/parquet-converter)<a href="https://agentmods.dev/skills/jdmorag97-rgb/ddc_skills_for_ai_agents_in_construction/parquet-converter"><img src="https://agentmods.dev/badge/skills/jdmorag97-rgb/ddc_skills_for_ai_agents_in_construction/parquet-converter/github.svg" alt="Measured on agentmods" height="20"></a>Or the 80×15 button, for a site that already has a row of RSS and ATOM ones. Only the verdict fits; the numbers stay here.
<a href="https://agentmods.dev/skills/jdmorag97-rgb/ddc_skills_for_ai_agents_in_construction/parquet-converter"><img src="https://agentmods.dev/badge/skills/jdmorag97-rgb/ddc_skills_for_ai_agents_in_construction/parquet-converter.svg" alt="Reviewed on agentmods" width="80" height="20"></a>What it costs to keep this loaded
Counted locally with the o200k_base tokenizer, which is exact for GPT models; Claude uses its own tokenizer and its counts differ. Treat this as one consistent yardstick across the catalogue rather than a bill. Prices are per million input tokens.
| Model | Per session | Once invoked |
|---|---|---|
| Fable 5.1 | $0.00028 | $0.03777 |
| Opus 5 | $0.00014 | $0.01888 |
| Sonnet 5 | $0.00006 | $0.00755 |
| Haiku 4.5 | $0.00003 | $0.00378 |
Grade A, and why
parquet-converter scanned grade A with 0 findings against 26 rules in 11 categories — prompt injection, anti-refusal, data exfiltration, privilege escalation, supply chain, agent snooping, system-prompt leakage, SSRF and excessive agency — measured 8d ago.
A static scan of the body, not an audit. Every finding is printed with the line that produced it so you can judge whether it matters here. A mod is markdown that instructs an agent; that is exactly why what it instructs is worth reading.
Nothing flagged
None of the 26 patterns this scan looks for appear in this file: no shell pipes, no recursive deletes, no credential paths, no hidden text, no instruction-override or anti-refusal phrasing, no agent-config snooping. That is not a guarantee, it is the absence of the things that are checkable.
This is a copy
100% identical to parquet-converter — 0 lines differ, which has more behind it and is treated as the original. This page carries a canonical link to it rather than competing with it.
How it starts
The opening of the file, as written. The whole thing — 504 lines — stays where its author put it; the contents beside it link to each section on GitHub.
Parquet Converter
Business Case
Problem Statement
Data storage and processing challenges:
- Large CSV files are slow to process
- Inefficient storage of typed data
- Column-oriented queries are slow
- Incompatible with modern data platforms
Solution
Convert construction data to Parquet format for efficient columnar storage, faster queries, and compatibility with data lakehouses.
Technical Implementation
import pandas as pd
from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
import json
class CompressionType:
SNAPPY = "snappy"
GZIP = "gzip"
BROTLI = "brotli"
ZSTD = "zstd"
NONE = None
@dataclass
class ParquetSchema:
columns: Dict[str, str] # column_name: dtype
partitions: List[str] = field(default_factory=list)
row_group_size: int = 100000
@dataclass
class ConversionResult:
source_path: str
output_path: str
source_format: str
rows: int
columns: int
original_size_mb: float
parquet_size_mb: float
compression_ratio: float
duration_seconds: float
class ParquetConverter:
"""Convert construction data to/from Parquet format."""
def __init__(self, project_name: str = "Data Conversion"):
self.project_name = project_name
self.conversions: List[ConversionResult] = []
self.schemas: Dict[str, ParquetSchema] = {}
self._define_standard_schemas()
def _define_standard_schemas(self):
"""Define standard schemas for construction data."""
self.schemas['projects'] = ParquetSchema(
columns={
'project_id': 'string',
'name': 'string',
'project_type': 'category',
'status': 'category',
'start_date': 'datetime64[ns]',
'end_date': 'datetime64[ns]',
'budget': 'float64',
'actual_cost': 'float64',
'size_sf': 'float64',
'location': 'string'
},
partitions=['project_type', 'status']
)
self.schemas['costs'] = ParquetSchema(
columns={
'transaction_id': 'string',
'project_id': 'string',
'cost_code': 'category',
'description': 'string',
'amount': 'float64',
'transaction_date': 'datetime64[ns]',
'vendor': 'string',
'invoice_number': 'string'
},
partitions=['project_id']
)
self.schemas['schedule'] = ParquetSchema(
columns={
'activity_id': 'string',
'project_id': 'string',
'name': 'string',
'wbs_code': 'string',
'start_date': 'datetime64[ns]',
'end_date': 'datetime64[ns]',
'duration': 'int32',
'progress': 'float32',
'status': 'category'
},
partitions=['project_id']
)
self.schemas['qto'] = ParquetSchema(
columns={
'element_id': 'string',
'project_id': 'string',
'element_type': 'category',
'name': 'string',
'quantity': 'float64',
'unit': 'category',
'level': 'string',
'material': 'string'
},
partitions=['project_id', 'element_type']
)
def add_schema(self, name: str, schema: ParquetSchema):
"""Add custom schema."""
self.schemas[name] = schema
def csv_to_parquet(self, csv_path: str, parquet_path: str,
schema_name: str = None,
compression: str = CompressionType.SNAPPY,
partition_cols: List[str] = None) -> ConversionResult:
"""Convert CSV to Parquet."""
start_time = datetime.now()
# Read CSV
df = pd.read_csv(csv_path)
# Apply schema if provided
if schema_name and schema_name in self.schemas:
schema = self.schemas[schema_name]
df = self._apply_schema(df, schema)
partition_cols = partition_cols or schema.partitions
# Get original file size
original_size = Path(csv_path).stat().st_size / (1024 * 1024)
# Write Parquet
if partition_cols:
# Partitioned write
available_partitions = [c for c in partition_cols if c in df.columns]
if available_partitions:
df.to_parquet(
parquet_path,
engine='pyarrow',
compression=compression,
partition_cols=available_partitions,
index=False
)
else:
df.to_parquet(parquet_path, engine='pyarrow',
compression=compression, index=False)
else:
df.to_parquet(parquet_path, engine='pyarrow',
compression=compression, index=False)
# Calculate parquet size
if Path(parquet_path).is_dir():
parquet_size = sum(f.stat().st_size for f in Path(parquet_path).rglob('*.parquet')) / (1024 * 1024)
else:
parquet_size = Path(parquet_path).stat().st_size / (1024 * 1024)
duration = (datetime.now() - start_time).total_seconds()
result = ConversionResult(
source_path=csv_path,
output_path=parquet_path,
source_format='csv',
rows=len(df),
columns=len(df.columns),
original_size_mb=round(original_size, 2),
parquet_size_mb=round(parquet_size, 2),
compression_ratio=round(original_size / parquet_size, 2) if parquet_size > 0 else 0,
duration_seconds=round(duration, 2)
)
self.conversions.append(result)
return result
def excel_to_parquet(self, excel_path: str, parquet_path: str,
sheet_name: Union[str, int] = 0,
schema_name: str = None,
compression: str = CompressionType.SNAPPY) -> ConversionResult:
"""Convert Excel to Parquet."""
start_time = datetime.now()
# Read Excel
df = pd.read_excel(excel_path, sheet_name=sheet_name)
# Apply schema
if schema_name and schema_name in self.schemas:
df = self._apply_schema(df, self.schemas[schema_name])
original_size = Path(excel_path).stat().st_size / (1024 * 1024)
# Write Parquet
df.to_parquet(parquet_path, engine='pyarrow',
compression=compression, index=False)
parquet_size = Path(parquet_path).stat().st_size / (1024 * 1024)
duration = (datetime.now() - start_time).total_seconds()
result = ConversionResult(
source_path=excel_path,
output_path=parquet_path,
source_format='excel',
rows=len(df),
columns=len(df.columns),
original_size_mb=round(original_size, 2),
parquet_size_mb=round(parquet_size, 2),
compression_ratio=round(original_size / parquet_size, 2) if parquet_size > 0 else 0,
duration_seconds=round(duration, 2)
)
self.conversions.append(result)
return result
def json_to_parquet(self, json_path: str, parquet_path: str,
schema_name: str = None,
compression: str = CompressionType.SNAPPY) -> ConversionResult:
"""Convert JSON to Parquet."""
start_time = datetime.now()
# Read JSON
df = pd.read_json(json_path)
if schema_name and schema_name in self.schemas:
df = self._apply_schema(df, self.schemas[schema_name])
original_size = Path(json_path).stat().st_size / (1024 * 1024)
df.to_parquet(parquet_path, engine='pyarrow',
compression=compression, index=False)
parquet_size = Path(parquet_path).stat().st_size / (1024 * 1024)
duration = (datetime.now() - start_time).total_seconds()
result = ConversionResult(
source_path=json_path,
output_path=parquet_path,
source_format='json',
rows=len(df),
columns=len(df.columns),
original_size_mb=round(original_size, 2),
parquet_size_mb=round(parquet_size, 2),
compression_ratio=round(original_size / parquet_size, 2) if parquet_size > 0 else 0,
duration_seconds=round(duration, 2)
)
self.conversions.append(result)
return result
def parquet_to_csv(self, parquet_path: str, csv_path: str) -> ConversionResult:
"""Convert Parquet to CSV."""
start_time = datetime.now()
df = pd.read_parquet(parquet_path)
if Path(parquet_path).is_dir():
original_size = sum(f.stat().st_size for f in Path(parquet_path).rglob('*.parquet')) / (1024 * 1024)
else:
original_size = Path(parquet_path).stat().st_size / (1024 * 1024)
df.to_csv(csv_path, index=False)
csv_size = Path(csv_path).stat().st_size / (1024 * 1024)
duration = (datetime.now() - start_time).total_seconds()
result = ConversionResult(
source_path=parquet_path,
output_path=csv_path,
source_format='parquet',
rows=len(df),
columns=len(df.columns),
original_size_mb=round(original_size, 2),
parquet_size_mb=round(csv_size, 2), # Actually CSV size
compression_ratio=round(csv_size / original_size, 2) if original_size > 0 else 0,
duration_seconds=round(duration, 2)
)
self.conversions.append(result)
return result
def _apply_schema(self, df: pd.DataFrame, schema: ParquetSchema) -> pd.DataFrame:
"""Apply schema to DataFrame."""
for col, dtype in schema.columns.items():
if col in df.columns:
try:
if dtype == 'category':
df[col] = df[col].astype('category')
elif dtype.startswith('datetime'):
df[col] = pd.to_datetime(df[col])
else:
df[col] = df[col].astype(dtype)
except (ValueError, TypeError):
pass # Keep original type if conversion fails
return df
def get_parquet_info(self, parquet_path: str) -> Dict[str, Any]:
"""Get information about a Parquet file."""
import pyarrow.parquet as pq
if Path(parquet_path).is_dir():
# Partitioned dataset
files = list(Path(parquet_path).rglob('*.parquet'))
total_size = sum(f.stat().st_size for f in files) / (1024 * 1024)
if files:
sample = pq.read_table(str(files[0]))
schema = sample.schema
else:
return {'error': 'No parquet files found'}
return {
'path': parquet_path,
'type': 'partitioned',
'num_files': len(files),
'total_size_mb': round(total_size, 2),
'columns': [f.name for f in schema],
'dtypes': {f.name: str(f.type) for f in schema}
}
else:
# Single file
pf = pq.ParquetFile(parquet_path)
metadata = pf.metadata
return {
'path': parquet_path,
'type': 'single_file',
'size_mb': round(Path(parquet_path).stat().st_size / (1024 * 1024), 2),
'num_rows': metadata.num_rows,
'num_columns': metadata.num_columns,
'num_row_groups': metadata.num_row_groups,
'columns': [pf.schema_arrow.field(i).name for i in range(metadata.num_columns)],
'created_by': metadata.created_by
}
def query_parquet(self, parquet_path: str, columns: List[str] = None,
filters: List[tuple] = None) -> pd.DataFrame:
"""Query Parquet file with column selection and filtering."""
return pd.read_parquet(
parquet_path,
columns=columns,
filters=filters
)
def merge_parquet_files(self, input_paths: List[str],
output_path: str,
compression: str = CompressionType.SNAPPY) -> ConversionResult:
"""Merge multiple Parquet files into one."""
start_time = datetime.now()
dfs = [pd.read_parquet(p) for p in input_paths]
merged = pd.concat(dfs, ignore_index=True)
original_size = sum(Path(p).stat().st_size for p in input_paths) / (1024 * 1024)
merged.to_parquet(output_path, engine='pyarrow',
compression=compression, index=False)
parquet_size = Path(output_path).stat().st_size / (1024 * 1024)
duration = (datetime.now() - start_time).total_seconds()
return ConversionResult(
source_path=str(input_paths),
output_path=output_path,
source_format='parquet_merge',
rows=len(merged),
columns=len(merged.columns),
original_size_mb=round(original_size, 2),
parquet_size_mb=round(parquet_size, 2),
compression_ratio=round(original_size / parquet_size, 2) if parquet_size > 0 else 0,
duration_seconds=round(duration, 2)
)
def get_conversion_summary(self) -> Dict[str, Any]:
"""Get summary of all conversions."""
if not self.conversions:
return {'total_conversions': 0}
return {
'total_conversions': len(self.conversions),
'total_rows_processed': sum(c.rows for c in self.conversions),
'original_size_mb': sum(c.original_size_mb for c in self.conversions),
'parquet_size_mb': sum(c.parquet_size_mb for c in self.conversions),
'avg_compression_ratio': round(
sum(c.compression_ratio for c in self.conversions) / len(self.conversions), 2
),
'total_duration_seconds': sum(c.duration_seconds for c in self.conversions)
}
def export_conversion_log(self, output_path: str) -> str:
"""Export conversion log to Excel."""
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Summary
summary = self.get_conversion_summary()
summary_df = pd.DataFrame([summary])
summary_df.to_excel(writer, sheet_name='Summary', index=False)
# Detailed log
log_df = pd.DataFrame([{
'Source': c.source_path,
'Output': c.output_path,
'Format': c.source_format,
'Rows': c.rows,
'Columns': c.columns,
'Original Size (MB)': c.original_size_mb,
'Parquet Size (MB)': c.parquet_size_mb,
'Compression Ratio': c.compression_ratio,
'Duration (s)': c.duration_seconds
} for c in self.conversions])
log_df.to_excel(writer, sheet_name='Conversions', index=False)
return output_path
What ships with it
2 files beside SKILL.md in the same directory: the scripts, references and assets a skill reads on demand. Not counted in the per-session cost; read them before you install if any of them is executable.
What this file has done since we first saw it
Hashed on every crawl. A supply-chain change to an agent config is a question of when, not whether, so the history is kept rather than the latest state alone.
- 8d ago First seen · 504 lines · 28 tokens per session scan A a6b69f32204b
parquet-converter is a skill published in the GitHub repository jdmorag97-rgb/DDC_Skills_for_AI_Agents_in_Construction (2 stars, last pushed 6mo ago), licensed MIT. It adds 28 tokens to every session and 3,777 once invoked, about $0.0001 per session on Opus 5. A static security scan graded it A with 0 findings. It is 100% identical to parquet-converter, differing in 0 lines, and is treated as a copy.
Other skills, from other repositories
pinecone
Managed vector DB for production RAG and search.
embeddings
Vector embeddings with HNSW indexing, sql.js persistence, and hyperbolic support. 75x faster with agentic-flow integration. Use when: semantic search, pattern matching, similarity queries, knowledge retrieval. Skip when: exact text matching, simple lookups, no semantic understanding needed.
data-engineer
Build scalable data pipelines, modern data warehouses, and real-time streaming architectures. Implements Apache Spark, dbt, Airflow, and cloud-native data platforms.
graphjin-env
Use when setting up a training or evaluation loop against a GraphJin agent environment — running the container, reading /health, driving episodes hosted or step-by-step or with your own agent over MCP, splitting train from eval, exporting trajectories, and deciding whether two rewards can be compared.
ingesting-into-data-lake
Import data into the AWS data lake from S3 files, local uploads, JDBC databases (Oracle, SQL Server, PostgreSQL, MySQL, RDS, Aurora), Amazon Redshift, Snowflake, BigQuery, DynamoDB, or existing Glue catalog tables (migration). Default target is S3 Tables; standard Iceberg on a general purpose bucket is supported where…
similarity-search-patterns
Implement efficient similarity search with vector databases. Use when building semantic search, implementing nearest neighbor queries, or optimizing retrieval performance.