[MicrosoftDataSchool] 33일차 - Azure IoT Hub

RudinP·2026년 2월 20일
MicrosoftDataSchoolazurestreamanalytics

Microsoft Data School 3기

목록 보기
34/51
post-thumbnail

RAG 사용자 입력을 sql query로 변환하는 부분 추가하기

직접 할 때 codex의 도움을 받으려 했으나, windows기반에서는 아직 불안정한듯 하여 mac으로 옮기느라 직접 작성해보지는 못했다.
강사님께서 짠 코드를 받아 추가하였다.

def text_to_sql(user_query):
    # Chat Completion API 호출
    utc_timestamp = datetime.datetime.now(tz=timezone.utc)
    kst = pytz.timezone("Asia/Seoul")
    kst_timestamp = utc_timestamp.astimezone(kst)
    base_date = kst_timestamp.strftime("%Y-%m-%d")
    
    # 이번 주 월요일과 수요일 날짜 계산
    monday_timestamp = kst_timestamp - timedelta(days=kst_timestamp.weekday())
    wednesday_timestamp = monday_timestamp + timedelta(days=2)

    monday_date = monday_timestamp.strftime("%Y-%m-%d")
    wednesday_date = wednesday_timestamp.strftime("%Y-%m-%d")

    # 어제 날짜 계산
    yesterday_timestamp = kst_timestamp - timedelta(days=1)
    yesterday_date = yesterday_timestamp.strftime("%Y-%m-%d")

    dynamic_examples_text = textwrap.dedent("""
    [참고 퓨샷(Few-shot) 예시]
    # 예시 1
    질문: 오늘 서울특별시 강남구의 데이터 조회해줘
    응답:
    {
        "reasoning": "사용자가 '오늘'(시스템 일자)의 '서울특별시'(city) 및 '강남구'(county) 데이터를 요청함. 단일 지역의 하루 조회를 의미하므로 intent_days=1, intent_regions=1, intent_hours=24로 추출. 직접적인 일치 조건(=) 필터링이 필요함.,
        "intent_analysis": {
            "intent_days": 1,
            "intent_regions": 1,
            "intent_hours": 24
        },
        "sql": "SELECT c.city, c.county, c.fcstDate, c.fcstTime, c.pcap, c.qgen, c.srad, c.temp, c.wspd FROM c WHERE c.city = @city AND c.county = @county AND c.fcstDate = @date",
        "parameters": [
            {"name": "@city", "value": "서울특별시"},
            {"name": "@county", "value": "강남구"},
            {"name": "@date", "value": "{base_date}"}
        ]
    }

    # 예시 2
    질문: 이번 주 월요일부터 수요일까지 경기도 지역 데이터 어떻게 돼?
    응답:
    {
        "reasoning": "기간(월~수, 즉 3일간)에 대한 범위 조회와 '경기도'(city) 필터링이 요구됨. 특정 시간 언급이 없으므로 intent_days=3, intent_regions=1, intent_hours=24로 추출. ISO 8601 문자열 특성을 활용하여 날짜 범위에 대해 시스템 함수 없이 BETWEEN 연산자를 사용함.",
        "intent_analysis": {
            "intent_days": 3,
            "intent_regions": 1,
            "intent_hours": 24
        },
        "sql": "SELECT c.city, c.county, c.fcstDate, c.fcstTime, c.pcap, c.qgen, c.srad, c.temp, c.wspd FROM c WHERE c.city = @city AND (c.fcstDate BETWEEN @start_date AND @end_date)",
        "parameters": [
            {"name": "@city", "value": "경기도"},
            {"name": "@start_date", "value": "{start_date}"},
            {"name": "@end_date", "value": "{end_date}"}
        ]
    }
                                            
    # 예시 3
    질문: 어제 오전 9시부터 오후 6시까지 평택시 데이터 알려줘
    응답:
    {
        "reasoning": "'어제'의 '평택시'(county) 데이터 중 09시부터 18시까지의 시간 범위 조회가 요구됨. 09시부터 18시까지는 총 10시간이므로 intent_days=1, intent_regions=1, intent_hours=10으로 추출. fcstTime 컬럼 역시 문자열이므로 BETWEEN 연산자를 적용함.",
        "intent_analysis": {
            "intent_days": 1,
            "intent_regions": 1,
            "intent_hours": 10
        },
        "sql": "SELECT c.city, c.county, c.fcstDate, c.fcstTime, c.pcap, c.qgen, c.srad, c.temp, c.wspd FROM c WHERE c.county = @county AND c.fcstDate = @date AND (c.fcstTime BETWEEN @start_time AND @end_time)",
        "parameters": [
            {"name": "@city", "value": "서울특별시"},
            {"name": "@county", "value": "강남구"},
            {"name": "@date", "value": "{yesterday_date}"},
            {"name": "@start_time", "value": "09"},
            {"name": "@end_time", "value": "18"},
        ]
    }
    """).replace("{base_date}", base_date).replace("{start_date}", monday_date).replace("{end_date}", wednesday_date).replace("{yesterday_date}", yesterday_date)

    messages = [
        {"role": "system", "content": textwrap.dedent("""
            - 현재 시스템 기준 일자: {base_date}
            - 대상 컬렉션(테이블) 별칭: c
            - 조회(SELECT) 결과에 반드시 포함해야 할 속성 목록: c.city, c.county, c.fcstDate, c.fcstTime, c.pcap, c.qgen, c.srad, c.temp, c.wspd
        
        [데이터 컬럼 정의]
        - city (String): 광역 자치단체 단위의 도시명 (예: 서울특별시, 경기도)
        - county (String): 기초 자치단체 단위의 시군구 (예: 강남구, 용인시)
        - fcstDate (String): 예측 일자. 반드시 'YYYY-MM-DD' ISO 8601 문자열 포맷을 유지해야 함.
        - fcstTime (STring): 예측 시간. 반드시 '01'부터 '23'까지의 두 자리 문자열 형식이어야 함.
        - 기타: pcap, qgen, srad, temp(기온), wspd(풍속) 등은 숫자형(Number) 데이터임.

        [데이터베이스 쿼리 작성 규칙 및 제약사항]
        1. 보안 최적화(SQL Injection 방어): 절대 WHERE 절 내에 검색 대상 리터럴 값을 직접 하드코딩하지 마세요. 사용자의 입력값은 무조건 '@' 기호로 시작하는 매개변수로 치환하여 `parameters` 배열에 분리해야 합니다.
        2. 인덱스 최적화(SARGability): fcstDate와 같은 문자열 날짜를 비교할 때, 'DateTimeToTimestamp'와 같은 시스템 변환 함수를 절대로 사용하지 마세요. Cosmos DB의 범위 인덱스 효율성을 극대화하기 위해, 원본 문자열 필드 자체를 부등호(>=, <=)나 BETWEEN 연산자로 직접 대소 비교해야 합니다.
        3. 비즈니스 논리 및 수치 연산 배제: 쿼리에 TOP N' 구문을 절대로 작성하거나 직접 수학 연산을 시도하지 마세요. 대신 사용자의 질문을 분석하여 며칠간의 데이터를 요구하는지(intent_days), 몇 개의 지역을 요구하는지(intent_regions), 몇 개의 시간인지(intent_hours)를 추출하여 `intent_analysis` 블록에 숫자로 제공하세요. 이를 바탕으로 애플리케이션 계층이 TOP 연산을 대신 수행합니다.
        4. 조건자 논리 무결성: BETWEEN 연산자나 여러 OR 조건이 다른 AND 조건과 결합될 경우, 논리적 문법 오류를 방지하기 위해 반드시 해당 부분 집합을 괄호 '()'로 명확히 묶어주세요.
        5. 스키마리스 속성 제어: 사용자가 시간(fcstTime)이나 시군구(county)를 구체적으로 지정하지 않았다면, 해당 조건을 WHERE 절에서 완전히 배제하여 불필요한 필터링을 막으세요. 만약 속성의 존재 여부 자체를 검사해야 할 상황이라면 'IS_DEFINED()' 함수를 활용하세요.

        {{
            "reasoning": "사용자의 질문 의도, 조건식 도출 과정, 날짜 비교 방식에 대한 1~2문장의 논리적 사고 과정",
            "intent_analysis": {{
                "intent_days": <Number: 요구된 인수 산출>,
                "intent_regions": <Number: 요구된 지역(도시/시군구) 개수 산출>,
                "intent_hours": <Number: 요구된 시간의 개수 산출. 하루 전체면 24, 특정 1시간이면 1, 범위형태면 해당 시간들의 개수>
            }},
            "sql": "작성된 매개변수화된 SELECT 쿼리 (TOP 구문 없이 작성)",
            "parameters": [
            {{"name": "@매개변수명", "value": "매핑할 문자열 혹은 숫자 값"}}
            ]
        }}

        {dynamic_examples_text}
    """).replace("{dynamic_examples_text}", dynamic_examples_text)},
    {"role": "user", "content": user_query}
    ]

    response = openai_client.chat.completions.create(
        model= openai_gpt_model,
        messages=messages,
        temperature=0.3
    )

    return response.choices[0].message.content


제대로 이틀간의 현황을 알려줄 수 있게 되었다.

참고로 sql injection을 시도하면 오류가 난다(근데 이건 gradio에서 json파싱 에러가 난것이라고 한다)

서울특별시' OR 1=1 -- 의 데이터를 조회해줘


자세한 로그는 함수에 들어가 모니터링-로그-traces에서 확인하면 된다.

보안 관련해서, ai 내에서 Guardrails + Controls에서 필터 규칙을 별도로 정할 수 있음을 참고하자(혐오 표현 방지 등)

Azure IoT Hub

사물 인터넷을 위한 클라우드 게이트웨이

event queue랑 보통 세트로 등장

IoT 솔루션의 핵심: 관리형 양방향 통신 서비스

중앙 메시지 허브

IoT 애플리케이션과 연결된 디바이스 간에 대규모로 통신을 수행하고 관리되는 클라우드 서비스

대규모 확장성

수백만 개의 동시 연결 디바이스와 초당 수백만 개의 이벤트를 처리

양방향 통신

디바이스 → 클라우드 데이터 전송 + 클라우드 → 디바이스 명령 및 제어 지원

네트워크 흐름

IoT 디바이스 프로파일

  • 무인 운영
  • 원격 및 열악한 환경에서 사용
  • 네트워크 제약(직접적 인터넷 연결 불가능)
  • 리소스 제한(전원 및 처리 리소스 제한적)
  • 간헐적 연결
  • 특수 프로토콜(MQTT, AMQP 등 산업별 독점(http안씀))

안전한 연결: ID 레지스트리 및 인증

모든 IoT Hub는 연결이 허용된 디바이스 정보를 저장하는 레지스트리를 보유
등록되지 않은 디바이스는 연결할 수 없음

SAS 토큰

  • 방식: 대칭 키
  • 특징: 구현 간단
  • 용도: 대칭 키를 안전하게 저장할 수 있는 디바이스에 적합

X.509 인증서

  • 방식: TLS 연결 중 디바이스 인증
  • 특징: 더 높은 보안 수준과 확장성 제공
  • 용도: 인증서 관리가 가능하고 높은 보안이 필요한 디바이스

대규모 디바이스 프로비저닝의 자동화

사람의 개입 없이 적절한 IoT Hub에 디바이스를 자동으로 프로비저닝
Just-in-Time등록을 통해 수백만개의 디바이스를 한 번에 설정하고 관리 가능

신뢰할 수 있는 양방향 통신

네트워크 중단이 발생하는 경우에도 메시지가 배달되도록 보장하는 신뢰할 수 있는 메시징 시스템

메시징 패턴 1: 원격 분석 및 속성

속성

디바이스 상태를 읽거나 설정(Device Twin/Plug & Play)

시나리오 A

Telemetry(Data Stream)
5분마다 IoT Hub로 현재 온도를 전송하는 냉동 트럭

시나리오 B

Properties(State Management)
배치 원자로를 모니터링하다가 온도가 특정 값을 초과할 때 경고를 보내는 화학 공장

메시징 패턴 2: 명령 및 제어

직접 메서드(Direct Methods)

HTTP 호출과 유사한 요청 회신(Request-Reply)voxjsdmf tkdyd
사용자가 지정한 시간 초과 후 즉시 성공 또는 실패를 반환

예시: 관리자가 원격으로 디바이스 재부팅 명령 실행 → 디바이스가 즉시 재부팅 후 성공 메시지 반환

데이터 수집 및 보존

Ingestion

  • 디바이스-클라우드 메시지
  • 클라우드-디바이스 메시지
  • 파일 업로드 시작
  • 디바이스 쌍(Twin) 속성
  • 직접 메서드 요청

Azure IoT Hub 실습 사전준비

Azure IoT Explorer 설치 파일 다운로드

Azure IoT Explorer는 Azure IoT 디바이스와 IoT Hub를 쉽게 관리하고 데이터 모니터링, 제어할 수 있는 사용자 친화적 그래픽 인터페이스 툴

설치 링크

참고로 맥은 안된다고 한다. 그래서 그냥 난 Azure Cli 깔고 IoT 확장을 설치했다.

# 1.Azure CLI 설치
brew install azure-cli
# 2.IoT 확장 설치
az extension add --name azure-iot
# 3. 로그인
az login

Azure Portal을 사용하여 IoT Hub 생성

Azure IoT Hub로 Device를 등록하고 원격 분석 전송

리소스 생성

IoT Hub 생성

  1. 만들기 버튼 클릭

  2. IoT Hub 인스턴스 정보 입력

  3. 검토+만들기

디바이스 등록

IoT Hub에 접속할 수 있는 마스터 키를 얻고 디바이스 등록하기

기본 문자열 가져오기

  1. 공유 액세스 정책 클릭
  2. iothubowner 클릭
  3. 기본 연결 문자열 복사

cli로 연결하기

az iot hub device-identity create \
  --login '연결기본문자열' \
  --device-id [디바이스이름]

cli로 연결 확인하기

az iot hub device-identity show \
  --login "연결문자열" \
  --device-id [디바이스이름]

환경변수로 연결기본문자열 설정하기

export IOTHUB_LOGIN='연결 문자열'

영구적으로(현재 콘솔 껐다 켜도) 쓰려면
nano ~/.zshrc 후 맨 아랫줄에 위 명령어 입력 후 저장

이후 사용

전체 디바이스 목록 확인 명령어
az iot hub device-identity list --login "$IOTHUB_LOGIN"

az iot hub device-identity show --login "$IOTHUB_LOGIN" --device-id mydevice

가상 디바이스 실행

가상의 온도 조절기 디바이스 시뮬레이션

가상환경 추가

python -m venv .venv
source .₩.venv₩bin₩activate
python -m pip install azure-iot-device

IoT Hub 연결 문자열 환경 변수 설정하기

  1. .env 파일 생성
  2. 값 입력
IOTHUB_DEVICE_CONNECTION_STRING="Primary connection string값"
IOTHUB_DEVICE_SECURITY_TYPE="connectionString"

CLI에서 디바이스 연결 문자열 뽑기

az iot hub device-identity connection-string show \
  --login $IOTHUB_LOGIN \
  --device-id mydevice \
  -o tsv

환경 변수를 읽는 기능 추가하기

  1. temp_controller_with_thermostats.py 파일 클릭
  2. temp_controller_with_thermostats.py의 10번째 줄에 from dotenv import load_dotenv 추가
  3. temp_controller_with_thermostats.py의 18번째 줄에 load_dotenv() 추가

가상 온도 데이터 실시간 확인

vscode python 결과 콘솔

모니터링

az iot hub monitor-events \
  --login "$IOTHUB_LOGIN"
  
특정 디바이스 모니터링
az iot hub monitor-events \
  --hub-name [IoT허브이름] \
  --device-id [디바이스이름]

profile
RudinP
iOS 개발자가 되기 위한 스터디룸/스터디의 레퍼런스는 모두 kxcoding
이전 포스트

[MicrosoftDataSchool] 32일차 - RAG 시스템 구축 및 Gradio를 이용한 챗봇 생성

다음 포스트

[MicrosoftDataSchool] 34일차 - 오픈소스를 활용한 데이터 파이프라인 구축

0개의 댓글