Fivetran QVD Connector

A custom Fivetran connector that reads Qlik QVD files and loads them into destinations via the Fivetran Connector SDK.

Features

• QVD File Support: Reads .qvd files from local filesystem or S3
• Incremental Sync: Supports mtime and mtime+sha1 strategies for efficient incremental updates
• Schema Inference: Automatically maps QVD field types to Fivetran SDK types
• Chunked Processing: Handles large files efficiently with configurable chunk sizes
• Primary Key Detection: Automatically detects ID columns or allows manual configuration
• Fallback Reader: Uses qvd (Rust-backed) with fallback to pyqvd (pure Python)
• Pattern Filtering: Include/exclude file patterns for selective processing
• Docker Support: Production-ready containerized deployment

Quick Start

Local Development

1. Install dependencies:

   pip install -r requirements.txt
   

2. Set environment variables:

   export QVD_SOURCE=local
   export QVD_DIR=/path/to/your/qvd/files
   export LOG_LEVEL=INFO
   

3. Run the connector:

   python connector.py
   

Docker Deployment

1. Build the image:

   docker build -t fivetran-qvd-connector .
   

2. Run with configuration:

   docker run -e QVD_SOURCE=local \
              -e QVD_DIR=/data/qvd \
              -v /path/to/qvd/files:/data/qvd \
              fivetran-qvd-connector
   

Configuration

Environment Variables

Variable	Default	Description
QVD_SOURCE	local	Source type: local or s3
QVD_DIR	/data/qvd	Local directory path for QVD files
S3_BUCKET	-	S3 bucket name (required for S3 mode)
S3_PREFIX	``	S3 prefix for file filtering
AWS_ACCESS_KEY_ID	-	AWS access key (optional)
AWS_SECRET_ACCESS_KEY	-	AWS secret key (optional)
AWS_REGION	us-east-1	AWS region
TABLE_NAME_MODE	file_stem	Table naming: file_stem or directory_file
CHUNK_SIZE_ROWS	50000	Rows per chunk for processing
STATE_STRATEGY	mtime	Incremental strategy: mtime or mtime+sha1
INCLUDE_PATTERNS	*.qvd	Comma-separated include patterns
EXCLUDE_PATTERNS	``	Comma-separated exclude patterns
FORCE_PK	-	Force primary key columns (comma-separated)
LOG_LEVEL	INFO	Logging level: DEBUG, INFO, WARNING, ERROR

Examples

Local files with custom patterns:

export QVD_SOURCE=local
export QVD_DIR=/data/qvd
export INCLUDE_PATTERNS="sales_*.qvd,customer_*.qvd"
export EXCLUDE_PATTERNS="temp_*,backup_*"
export TABLE_NAME_MODE=directory_file

S3 with authentication:

export QVD_SOURCE=s3
export S3_BUCKET=my-qvd-bucket
export S3_PREFIX=data/
export AWS_ACCESS_KEY_ID=your_key
export AWS_SECRET_ACCESS_KEY=your_secret
export AWS_REGION=us-west-2

High-performance configuration:

export CHUNK_SIZE_ROWS=100000
export STATE_STRATEGY=mtime+sha1
export FORCE_PK=id,created_at

File Structure

Fivetran-QVD-Connector/
├── connector.py          # Main entrypoint
├── discovery.py          # File/S3 discovery
├── qvd_reader.py        # QVD reading with fallback
├── schema.py            # Type mapping & sanitization
├── state.py             # Incremental sync state
├── utils.py             # Configuration & utilities
├── requirements.txt     # Python dependencies
├── Dockerfile           # Container definition
├── tests/               # Test suite
│   └── test_connector.py
└── README.md            # This file

Schema Mapping

QVD field types are automatically mapped to Fivetran SDK types:

QVD Type	Fivetran Type	Notes
integer, int, long	INT	64-bit integers
real, float, double, numeric	FLOAT	Floating point numbers
string, text, varchar, char	STRING	Text data
boolean, bool	BOOLEAN	True/false values
date, datetime, timestamp, time	DATETIME	Date/time values

Primary Key Detection

The connector automatically detects primary keys using these heuristics:

1. Columns named id (case-insensitive)
2. Columns ending with _id (e.g., user_id, order_id)
3. Columns starting with pk_ or key_

You can override this with the FORCE_PK environment variable.

Incremental Sync Strategies

MTime Strategy (Default)

• Uses file modification time
• Fastest option
• May miss content changes that don't update mtime

MTime + SHA1 Strategy

• Uses both modification time and file hash
• More reliable for detecting changes
• Slightly slower due to hash computation

Error Handling

The connector includes comprehensive error handling:

• Retry Logic: Transient I/O errors are retried with exponential backoff
• Graceful Degradation: Individual file failures don't stop the entire sync
• Validation: Files are validated before processing
• Logging: Structured logging for debugging and monitoring

Testing

Run the test suite:

pytest tests/

The tests include:

• Schema mapping and validation
• State management
• File discovery patterns
• QVD reader fallback logic
• Integration scenarios

Performance Considerations

• Chunk Size: Larger chunks reduce API calls but increase memory usage
• Reader Choice: qvd (Rust) is faster than pyqvd (Python) for large files
• State Strategy: mtime is faster than mtime+sha1 but less reliable
• S3 Mode: Files are downloaded temporarily, consider disk space

Troubleshooting

Common Issues

"No QVD reader available":

• Install qvd or pyqvd: pip install qvd or pip install pyqvd

"S3 access denied":

• Check AWS credentials and bucket permissions
• Verify S3 bucket and prefix configuration

"Invalid QVD file":

• Ensure file is a valid QVD format
• Check file permissions and accessibility

"Schema validation failed":

• Check for duplicate column names
• Verify primary key columns exist in schema

Debugging

Enable debug logging:

export LOG_LEVEL=DEBUG

This will show detailed information about:

• File discovery process
• Schema inference
• Row processing chunks
• State updates

Contributing

1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Ensure all tests pass
5. Submit a pull request

License

This connector is provided as-is for use with Fivetran's Connector SDK.