مهندس البيانات

فيديوهات تعليمية

ما هو مهندس البيانات؟

مهندس البيانات

مهندس البيانات هو متخصص يصمم ويبني أنظمة جمع البيانات وتخزينها ومعالجتها، مما يمكن الشركات من تحليل كميات كبيرة من البيانات بكفاءة.

تخزين البيانات

تصميم أنظمة تخزين البيانات بكفاءة

معالجة البيانات

تحويل وتنظيف البيانات للتحليل

ETL Pipelines

بناء خطوط أنابيب استخراج وتحويل وتحميل

معالجة الوقت الفعلي

معالجة البيانات المباشرة والتدفقات

اللغات والأدوات المستخدمة

Python

اللغة الأساسية للتعامل مع البيانات وتحليل النماذج

SQL

لإدارة قواعد البيانات المحلية واستعلام البيانات

Java/Scala

(اختياري) عند العمل مع Hadoop و Apache Spark

Shell Scripting

لأتمتة المهام في أنظمة Unix/Linux

Docker

لتوفير بيئة موحدة وأتمتة النشر

Git/GitHub

لإدارة النسخ البرمجية والتعاون

مهارات مهندس البيانات

Python Programming

إتقان برمجة Python للتعامل مع البيانات

SQL & Databases

إدارة قواعد البيانات واستعلامات SQL

Apache Spark

معالجة البيانات الضخمة بسرعة

ETL Pipelines

بناء خطوط أنابيب استخراج وتحويل وتحميل

Data Warehousing

تصميم وتنفيذ مستودعات البيانات

Stream Processing

معالجة البيانات في الوقت الفعلي

مثال عملي Apache Kafka:

from kafka import KafkaProducer, KafkaConsumer
from kafka.admin import KafkaAdminClient, NewTopic
from kafka.errors import KafkaError
import json
import time
from datetime import datetime
from typing import Dict, List, Optional
import threading
import logging

# 1. إعداد التسجيل (Logging)
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 2. فئة لإدارة Kafka Producer
class DataStreamProducer:
    """منتج تدفق البيانات إلى Kafka"""
    
    def __init__(self, bootstrap_servers: List[str]):
        self.bootstrap_servers = bootstrap_servers
        self.producer = None
        self._connect()
    
    def _connect(self):
        """الاتصال بـ Kafka"""
        try:
            self.producer = KafkaProducer(
                bootstrap_servers=self.bootstrap_servers,
                value_serializer=lambda v: json.dumps(v).encode('utf-8'),
                key_serializer=lambda k: str(k).encode('utf-8') if k else None,
                acks='all',
                retries=3,
                compression_type='gzip',
                max_in_flight_requests_per_connection=1
            )
            logger.info("تم الاتصال بـ Kafka Producer بنجاح")
        except Exception as e:
            logger.error(f"فشل الاتصال بـ Kafka: {e}")
            raise
    
    def produce_message(self, topic: str, key: Optional[str], value: Dict):
        """إنتاج رسالة إلى Kafka"""
        try:
            future = self.producer.send(
                topic=topic,
                key=key,
                value=value
            )
            
            # انتظار التأكيد
            result = future.get(timeout=10)
            logger.info(f"تم إرسال الرسالة إلى {topic} - Partition: {result.partition}, Offset: {result.offset}")
            return True
        except KafkaError as e:
            logger.error(f"فشل إرسال الرسالة: {e}")
            return False
    
    def produce_batch(self, topic: str, messages: List[Dict]):
        """إنتاج دفعة من الرسائل"""
        success_count = 0
        fail_count = 0
        
        for i, message in enumerate(messages):
            key = message.get('id', str(i))
            if self.produce_message(topic, key, message):
                success_count += 1
            else:
                fail_count += 1
            
            # إضافة تأخير صغير بين الرسائل
            time.sleep(0.01)
        
        logger.info(f"الدفعة: {success_count} ناجحة، {fail_count} فاشلة")
        return success_count, fail_count
    
    def close(self):
        """إغلاق المنتج"""
        if self.producer:
            self.producer.flush()
            self.producer.close()
            logger.info("تم إغلاق Kafka Producer")

# 3. فئة لإدارة Kafka Consumer
class DataStreamConsumer:
    """مستهلك تدفق البيانات من Kafka"""
    
    def __init__(self, bootstrap_servers: List[str], group_id: str):
        self.bootstrap_servers = bootstrap_servers
        self.group_id = group_id
        self.consumer = None
        self.running = False
        self._connect()
    
    def _connect(self):
        """الاتصال بـ Kafka"""
        try:
            self.consumer = KafkaConsumer(
                bootstrap_servers=self.bootstrap_servers,
                group_id=self.group_id,
                value_deserializer=lambda v: json.loads(v.decode('utf-8')),
                key_deserializer=lambda k: k.decode('utf-8') if k else None,
                auto_offset_reset='earliest',
                enable_auto_commit=True,
                auto_commit_interval_ms=1000,
                max_poll_records=100,
                session_timeout_ms=30000,
                heartbeat_interval_ms=10000
            )
            logger.info("تم الاتصال بـ Kafka Consumer بنجاح")
        except Exception as e:
            logger.error(f"فشل الاتصال بـ Kafka: {e}")
            raise
    
    def subscribe(self, topics: List[str]):
        """الاشتراك في مواضيع"""
        self.consumer.subscribe(topics)
        logger.info(f"تم الاشتراك في المواضيع: {topics}")
    
    def consume_messages(self, callback):
        """استهلاك الرسائل"""
        self.running = True
        
        try:
            while self.running:
                # جلب الرسائل
                batch = self.consumer.poll(timeout_ms=1000)
                
                for topic_partition, messages in batch.items():
                    logger.info(f"جلب {len(messages)} رسالة من {topic_partition}")
                    
                    for message in messages:
                        try:
                            # معالجة الرسالة
                            result = callback({
                                'topic': topic_partition.topic,
                                'partition': topic_partition.partition,
                                'offset': message.offset,
                                'key': message.key,
                                'value': message.value,
                                'timestamp': message.timestamp
                            })
                            
                            if result is False:
                                logger.warning("توقف المعالجة بناءً على رد الدالة")
                                return
                        
                        except Exception as e:
                            logger.error(f"خطأ في معالجة الرسالة: {e}")
                
                # التزام بالإزاحة
                self.consumer.commit()
        
        except KeyboardInterrupt:
            logger.info("توقف المستهلك بواسطة المستخدم")
        except Exception as e:
            logger.error(f"خطأ في استهلاك الرسائل: {e}")
        finally:
            self.close()
    
    def consume_single_message(self, timeout_ms: int = 10000):
        """استهلاك رسالة واحدة"""
        try:
            messages = self.consumer.poll(timeout_ms=timeout_ms)
            
            for topic_partition, msgs in messages.items():
                if msgs:
                    message = msgs[0]
                    return {
                        'topic': topic_partition.topic,
                        'partition': topic_partition.partition,
                        'offset': message.offset,
                        'key': message.key,
                        'value': message.value,
                        'timestamp': message.timestamp
                    }
        
        except Exception as e:
            logger.error(f"خطأ في استهلاك الرسالة: {e}")
        
        return None
    
    def close(self):
        """إغلاق المستهلك"""
        if self.consumer:
            self.consumer.close()
            logger.info("تم إغلاق Kafka Consumer")

# 4. فئة لإدارة Kafka Topics
class KafkaTopicManager:
    """مدير مواضيع Kafka"""
    
    def __init__(self, bootstrap_servers: List[str]):
        self.bootstrap_servers = bootstrap_servers
        self.admin_client = None
        self._connect()
    
    def _connect(self):
        """الاتصال بـ Kafka Admin"""
        try:
            self.admin_client = KafkaAdminClient(
                bootstrap_servers=self.bootstrap_servers,
                client_id='kafka_topic_manager'
            )
            logger.info("تم الاتصال بـ Kafka Admin Client بنجاح")
        except Exception as e:
            logger.error(f"فشل الاتصال بـ Kafka Admin: {e}")
            raise
    
    def create_topic(self, topic_name: str, num_partitions: int = 3, replication_factor: int = 1):
        """إنشاء موضوع جديد"""
        try:
            topic = NewTopic(
                name=topic_name,
                num_partitions=num_partitions,
                replication_factor=replication_factor,
                topic_configs={
                    'retention.ms': '604800000',  # 7 أيام
                    'cleanup.policy': 'delete',
                    'compression.type': 'gzip'
                }
            )
            
            self.admin_client.create_topics([topic])
            logger.info(f"تم إنشاء الموضوع: {topic_name}")
            return True
        
        except Exception as e:
            logger.error(f"فشل إنشاء الموضوع {topic_name}: {e}")
            return False
    
    def delete_topic(self, topic_name: str):
        """حذف موضوع"""
        try:
            self.admin_client.delete_topics([topic_name])
            logger.info(f"تم حذف الموضوع: {topic_name}")
            return True
        
        except Exception as e:
            logger.error(f"فشل حذف الموضوع {topic_name}: {e}")
            return False
    
    def list_topics(self):
        """عرض جميع المواضيع"""
        try:
            topics = self.admin_client.list_topics()
            logger.info(f"المواضيع المتاحة: {topics}")
            return topics
        
        except Exception as e:
            logger.error(f"فشل عرض المواضيع: {e}")
            return []
    
    def describe_topic(self, topic_name: str):
        """وصف موضوع"""
        try:
            configs = self.admin_client.describe_configs(
                config_resources=[ConfigResource(ConfigResourceType.TOPIC, topic_name)]
            )
            
            for config_resource in configs:
                logger.info(f"إعدادات الموضوع {topic_name}:")
                for key, value in config_resource.items():
                    logger.info(f"  {key}: {value.value}")
            
            return configs
        
        except Exception as e:
            logger.error(f"فشل وصف الموضوع {topic_name}: {e}")
            return None
    
    def close(self):
        """إغلاق اتصال Admin"""
        if self.admin_client:
            self.admin_client.close()
            logger.info("تم إغلاق Kafka Admin Client")

# 5. فئة لمعالجة تدفق البيانات في الوقت الفعلي
class RealTimeDataProcessor:
    """معالج بيانات في الوقت الفعلي باستخدام Kafka"""
    
    def __init__(self, bootstrap_servers: List[str], source_topic: str, sink_topic: str):
        self.bootstrap_servers = bootstrap_servers
        self.source_topic = source_topic
        self.sink_topic = sink_topic
        
        self.producer = DataStreamProducer(bootstrap_servers)
        self.consumer = DataStreamConsumer(bootstrap_servers, 'realtime_processor_group')
        self.topic_manager = KafkaTopicManager(bootstrap_servers)
    
    def setup_topics(self):
        """إعداد المواضيع"""
        # إنشاء مواضيع إذا لم تكن موجودة
        topics = self.topic_manager.list_topics()
        
        if self.source_topic not in topics:
            self.topic_manager.create_topic(self.source_topic)
        
        if self.sink_topic not in topics:
            self.topic_manager.create_topic(self.sink_topic)
    
    def process_message(self, message: Dict) -> Optional[Dict]:
        """معالجة رسالة فردية"""
        try:
            raw_data = message['value']
            
            # تحويل البيانات
            processed_data = {
                'id': raw_data.get('id', message['key']),
                'timestamp': datetime.now().isoformat(),
                'source_topic': message['topic'],
                'source_partition': message['partition'],
                'source_offset': message['offset'],
                'customer_id': raw_data.get('customer_id'),
                'event_type': raw_data.get('event_type'),
                'amount': raw_data.get('amount', 0),
                'location': raw_data.get('location', {}),
                'metadata': {
                    'processing_time': datetime.now().isoformat(),
                    'processor_version': '1.0.0'
                }
            }
            
            # تحويلات إضافية
            if 'amount' in raw_data:
                processed_data['amount_usd'] = raw_data['amount'] * 3.75  # تحويل إلى دولار
                processed_data['amount_category'] = self._categorize_amount(raw_data['amount'])
            
            if 'timestamp' in raw_data:
                processed_data['hour_of_day'] = datetime.fromisoformat(raw_data['timestamp']).hour
                processed_data['day_of_week'] = datetime.fromisoformat(raw_data['timestamp']).strftime('%A')
            
            # فحص الجودة
            if self._validate_data(processed_data):
                return processed_data
            else:
                logger.warning(f"بيانات غير صالحة: {processed_data['id']}")
                return None
        
        except Exception as e:
            logger.error(f"خطأ في معالجة الرسالة: {e}")
            return None
    
    def _categorize_amount(self, amount: float) -> str:
        """تصنيف المبلغ"""
        if amount < 100:
            return 'صغير'
        elif amount < 1000:
            return 'متوسط'
        else:
            return 'كبير'
    
    def _validate_data(self, data: Dict) -> bool:
        """التحقق من صحة البيانات"""
        required_fields = ['id', 'customer_id', 'event_type']
        
        for field in required_fields:
            if not data.get(field):
                return False
        
        if data.get('amount') and data['amount'] < 0:
            return False
        
        return True
    
    def start_processing(self):
        """بدء المعالجة"""
        logger.info("بدء معالجة البيانات في الوقت الفعلي")
        
        # إعداد المواضيع
        self.setup_topics()
        
        # الاشتراك في الموضوع المصدر
        self.consumer.subscribe([self.source_topic])
        
        # دالة رد الاتصال للمعالجة
        def process_callback(message):
            # معالجة الرسالة
            processed_data = self.process_message(message)
            
            if processed_data:
                # إرسال البيانات المعالجة إلى الموضوع الهدف
                success = self.producer.produce_message(
                    topic=self.sink_topic,
                    key=processed_data['id'],
                    value=processed_data
                )
                
                if success:
                    logger.info(f"تم معالجة وإرسال: {processed_data['id']}")
                else:
                    logger.error(f"فشل إرسال: {processed_data['id']}")
            
            return True  # مواصلة المعالجة
        
        # بدء استهلاك البيانات
        self.consumer.consume_messages(process_callback)
    
    def stop_processing(self):
        """إيقاف المعالجة"""
        logger.info("إيقاف معالجة البيانات")
        self.consumer.running = False
    
    def close(self):
        """إغلاق جميع الاتصالات"""
        self.producer.close()
        self.consumer.close()
        self.topic_manager.close()
        logger.info("تم إغلاق جميع اتصالات Kafka")

# 6. مثال استخدام
if __name__ == "__main__":
    # تكوين Kafka
    BOOTSTRAP_SERVERS = ['localhost:9092']
    SOURCE_TOPIC = 'raw_events'
    SINK_TOPIC = 'processed_events'
    
    # إنشاء معالج البيانات
    processor = RealTimeDataProcessor(BOOTSTRAP_SERVERS, SOURCE_TOPIC, SINK_TOPIC)
    
    try:
        # بدء المعالجة في thread منفصل
        processing_thread = threading.Thread(target=processor.start_processing)
        processing_thread.start()
        
        # إنتاج بعض البيانات التجريبية
        producer = DataStreamProducer(BOOTSTRAP_SERVERS)
        
        # بيانات تجريبية
        test_messages = [
            {
                'id': 'event_001',
                'customer_id': 'cust_123',
                'event_type': 'purchase',
                'amount': 150.50,
                'timestamp': datetime.now().isoformat(),
                'location': {'city': 'الرياض', 'country': 'السعودية'}
            },
            {
                'id': 'event_002',
                'customer_id': 'cust_456',
                'event_type': 'view',
                'amount': 0,
                'timestamp': datetime.now().isoformat(),
                'location': {'city': 'جدة', 'country': 'السعودية'}
            },
            {
                'id': 'event_003',
                'customer_id': 'cust_789',
                'event_type': 'purchase',
                'amount': 2500.00,
                'timestamp': datetime.now().isoformat(),
                'location': {'city': 'الدمام', 'country': 'السعودية'}
            }
        ]
        
        # إرسال البيانات
        for message in test_messages:
            producer.produce_message(SOURCE_TOPIC, message['id'], message)
            time.sleep(1)
        
        # انتظار قليل لمعالجة البيانات
        time.sleep(5)
        
        # استهلاك البيانات المعالجة
        consumer = DataStreamConsumer(BOOTSTRAP_SERVERS, 'test_consumer_group')
        consumer.subscribe([SINK_TOPIC])
        
        print("📥 البيانات المعالجة:")
        for _ in range(3):
            processed_message = consumer.consume_single_message(5000)
            if processed_message:
                print(json.dumps(processed_message['value'], indent=2, ensure_ascii=False))
        
        consumer.close()
        producer.close()
        
    except KeyboardInterrupt:
        print("\nإيقاف المعالجة...")
    finally:
        processor.stop_processing()
        processor.close()

# 7. التكامل مع Spark Streaming
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

def create_spark_kafka_stream():
    """إنشاء تدفق Spark من Kafka"""
    spark = SparkSession.builder \
        .appName("KafkaSparkIntegration") \
        .config("spark.sql.shuffle.partitions", "100") \
        .getOrCreate()
    
    # قراءة من Kafka
    kafka_df = spark.readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "localhost:9092") \
        .option("subscribe", "processed_events") \
        .option("startingOffsets", "latest") \
        .load()
    
    # تعريف مخطط البيانات
    schema = StructType([
        StructField("id", StringType()),
        StructField("customer_id", StringType()),
        StructField("event_type", StringType()),
        StructField("amount", DoubleType()),
        StructField("amount_usd", DoubleType()),
        StructField("amount_category", StringType()),
        StructField("timestamp", TimestampType()),
        StructField("hour_of_day", IntegerType()),
        StructField("day_of_week", StringType()),
        StructField("location", StructType([
            StructField("city", StringType()),
            StructField("country", StringType())
        ])),
        StructField("metadata", StructType([
            StructField("processing_time", TimestampType()),
            StructField("processor_version", StringType())
        ]))
    ])
    
    # تحويل البيانات
    parsed_df = kafka_df \
        .select(from_json(col("value").cast("string"), schema).alias("data")) \
        .select("data.*")
    
    # معالجة التدفق
    processed_stream = parsed_df \
        .withWatermark("timestamp", "5 minutes") \
        .groupBy(
            window("timestamp", "1 hour"),
            "event_type",
            "amount_category"
        ).agg(
            count("*").alias("event_count"),
            sum("amount").alias("total_amount"),
            avg("amount").alias("avg_amount"),
            approx_count_distinct("customer_id").alias("unique_customers")
        )
    
    return processed_stream

# 8. مراقبة Kafka
class KafkaMonitor:
    """مراقب حالة Kafka"""
    
    def __init__(self, bootstrap_servers: List[str]):
        self.bootstrap_servers = bootstrap_servers
        self.admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)
    
    def get_cluster_health(self):
        """الحصول على حالة الكتلة"""
        try:
            # الحصول على معلومات الكتلة
            cluster_description = self.admin_client.describe_cluster()
            
            health_info = {
                'cluster_id': cluster_description.cluster_id,
                'controller_id': cluster_description.controller.id,
                'broker_count': len(cluster_description.brokers),
                'brokers': [
                    {
                        'id': broker.id,
                        'host': broker.host,
                        'port': broker.port,
                        'rack': broker.rack
                    }
                    for broker in cluster_description.brokers
                ]
            }
            
            return health_info
        
        except Exception as e:
            logger.error(f"خطأ في الحصول على حالة الكتلة: {e}")
            return None
    
    def get_topic_metrics(self, topic_name: str):
        """الحصول على مقاييس الموضوع"""
        try:
            # الحصول على أقسام الموضوع
            partitions = self.admin_client.list_topics().get(topic_name, [])
            
            metrics = {
                'topic': topic_name,
                'partition_count': len(partitions),
                'partitions': []
            }
            
            # هنا يمكن إضافة كود للحصول على مقاييس الأقسام
            # مثل حجم البيانات، الإزاحات، etc.
            
            return metrics
        
        except Exception as e:
            logger.error(f"خطأ في الحصول على مقاييس الموضوع: {e}")
            return None
    
    def close(self):
        """إغلاق الاتصال"""
        self.admin_client.close()

# 9. اختبارات Kafka
import unittest

class TestKafkaIntegration(unittest.TestCase):
    """اختبار تكامل Kafka"""
    
    def setUp(self):
        self.bootstrap_servers = ['localhost:9092']
        self.test_topic = 'test_topic'
    
    def test_producer_connection(self):
        """اختبار اتصال المنتج"""
        producer = DataStreamProducer(self.bootstrap_servers)
        self.assertIsNotNone(producer.producer)
        producer.close()
    
    def test_consumer_connection(self):
        """اختبار اتصال المستهلك"""
        consumer = DataStreamConsumer(self.bootstrap_servers, 'test_group')
        self.assertIsNotNone(consumer.consumer)
        consumer.close()
    
    def test_topic_creation(self):
        """اختبار إنشاء الموضوع"""
        topic_manager = KafkaTopicManager(self.bootstrap_servers)
        success = topic_manager.create_topic(self.test_topic)
        self.assertTrue(success)
        
        # التحقق من وجود الموضوع
        topics = topic_manager.list_topics()
        self.assertIn(self.test_topic, topics)
        
        topic_manager.close()

if __name__ == '__main__':
    unittest.main()

هندسة أنظمة البيانات

استخراج البيانات

APIs، قواعد بيانات، ملفات

معالجة البيانات

ETL/ELT، التنظيف، التحويل

تخزين البيانات

Data Warehouses، Data Lakes

التحليل

BI Tools، التحليلات، التقارير

أدوات هندسة البيانات

قواعد البيانات

PostgreSQL، MySQL، MongoDB، Cassandra

SQL NoSQL NewSQL

معالجة البيانات

Apache Spark، Apache Flink، Apache Beam

Batch Streaming Real-time

إدارة سير العمل

Apache Airflow، Luigi، Prefect

Orchestration Scheduling Monitoring

المزايا والتحديات

المزايا

طلب عالي: هناك طلب كبير على مهندسي البيانات، خاصة في الشركات التي تعتمد على البيانات الضخمة
أدوات مجانية: معظم الأدوات المستخدمة مثل Apache Spark و Hadoop مجانية ومفتوحة المصدر
مجتمع كبير: Python و Apache Spark لديهما مجتمعات نشطة توفر الدعم والموارد
إبداع لا محدود: يمكنك بناء أنظمة معقدة لمعالجة البيانات وتحليلها
رواتب ممتازة: مهندسو البيانات من أعلى المهن دخلاً في مجال التكنولوجيا

التحديات

منحنى التعلم الحاد: يتطلب فهماً جيداً لـ Python والرياضيات والإحصاء
حجم البيانات: قد تواجه تحديات في التعامل مع مجموعات بيانات كبيرة جداً
تحديثات متكررة: الأدوات والتقنيات تتطور باستمرار، مما يتطلب تحديث المعرفة بشكل منتظم
تعقيد الأنظمة: أنظمة هندسة البيانات معقدة وتتطلب فهماً عميقاً للهندسة

تخصصات في هندسة البيانات

هندسة ETL

تصميم وتنفيذ خطوط أنابيب استخراج وتحويل وتحميل

Airflow Spark SQL

هندسة التدفقات

معالجة البيانات في الوقت الفعلي والتدفقات

Kafka Flink Spark

هندسة المستودعات

تصميم وتنفيذ مستودعات البيانات

BigQuery Redshift Snowflake

الخلاصة

مهندس البيانات مجال متقدم ومطلوب بشدة في سوق العمل. من خلال إتقان Python، SQL، Apache Spark، Hadoop، Airflow، و Kafka، يمكنك بناء أنظمة متكاملة لجمع ومعالجة وتحليل البيانات بكفاءة عالية.

نصائح للبدء:

ابدأ بإتقان Python و SQL كأساس
تعلم Apache Spark لمعالجة البيانات الضخمة
اتقن إدارة قواعد البيانات SQL و NoSQL
تعلم Airflow لأتمتة سير العمل
جرب Kafka لمعالجة البيانات في الوقت الفعلي

موارد تعليمية:

Apache Spark Apache Airflow Apache Kafka Apache Hadoop