dynamicDagfrom datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from db_utils import execute_query import logging def log_task_start(message): logging.info(f"[START] {message}") def log_task_end(message): logging.info(f"[END] {message}") # 분석할 주식 종목 리스트 (설정으로 관리) STOCK_SYMBOLS = ['AAPL','GOOGL','MSFT','AMZN','TSLA'] default_args = { 'owner': 'data-team', 'depends_on_past': False, 'start_date': datetime(2024, 1, 1), 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), } def analyze_stock_data(symbol, **context): """주식 종목에 대해서, 최근 30일간의 주식 데이터를 분석""" log_task_start(f"주식 분석 - {symbol}") query = """ SELECT symbol, AVG(price) as avg_price, MAX(price) as max_price, MIN(price) as min_price, SUM(volume) as total_volume, COUNT(*) as trading_days FROM stock_prices WHERE symbol = %s AND trade_date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) GROUP BY symbol """ result = execute_query(query, [symbol]) if result: data = result[0] print(f"{symbol} 분석 결과:") print(f" 평균 가격: ${data[1]:.2f}") print(f" 최고 가격: ${data[2]:.2f}") print(f" 최저 가격: ${data[3]:.2f}") print(f" 총 거래량: {data[4]:,}") print(f" 거래일수: {data[5]}일") log_task_end(f"주식 분석 - {symbol}") return result def create_stock_dag(symbol): """개별 주식에 대한 DAG 생성 함수""" dag_id = f'stock_analysis_{symbol.lower()}' dag = DAG( dag_id, default_args=default_args, description=f'{symbol} 주식 데이터 분석 DAG', schedule='@daily', catchup=False, tags=['dynamic', 'stock', 'analysis', symbol] ) # 데이터 품질 체크 태스크 def check_data_quality(**context): query = """ SELECT COUNT(*) as count FROM stock_prices WHERE symbol = %s AND trade_date = CURDATE() - INTERVAL 1 DAY """ result = execute_query(query, [symbol]) logging.info(f"{symbol} 데이터 품질 체크: {result[0][0] if result else 0}") return result data_quality_check = PythonOperator( task_id=f'check_data_quality_{symbol.lower()}', python_callable=check_data_quality, dag=dag ) # 주식 분석 태스크 analyze_task = PythonOperator( task_id=f'analyze_{symbol.lower()}', python_callable=analyze_stock_data, op_kwargs={'symbol': symbol}, dag=dag ) # 결과 저장 태스크 def save_results(**context): query = """ INSERT INTO daily_stats (symbol, summary_date, avg_price, total_volume, trade_count) SELECT %s, CURDATE() - INTERVAL 1 DAY, AVG(price), SUM(volume), COUNT(*) FROM stock_prices WHERE symbol = %s AND trade_date = CURDATE() - INTERVAL 1 DAY ON DUPLICATE KEY UPDATE avg_price = VALUES(avg_price), total_volume = VALUES(total_volume), trade_count = VALUES(trade_count) """ result = execute_query(query, [symbol, symbol]) logging.info(f"{symbol} 분석 결과 저장 완료") return result save_results = PythonOperator( task_id=f'save_analysis_{symbol.lower()}', python_callable=save_results, dag=dag ) # 태스크 의존성 설정 data_quality_check >> analyze_task >> save_results return dag # Dynamic DAG 생성 # 각 주식 종목에 대해 별도의 DAG 생성 for symbol in STOCK_SYMBOLS: dag_id = f'stock_analysis_{symbol.lower()}' globals()[dag_id] = create_stock_dag(symbol) CrossDag (트리거 센서)from datetime import datetime, timedelta from airflow import DAG, Dataset from airflow.operators.python import PythonOperator from airflow.operators.trigger_dagrun import TriggerDagRunOperator from airflow.sensors.external_task import ExternalTaskSensor from db_utils import execute_query import logging default_args = { 'owner': 'data-team', 'start_date': datetime(2024, 1, 1), 'retries': 1, 'retry_delay': timedelta(minutes=2), } def process_data(**context): """간단한 데이터 처리""" logging.info("데이터 처리 시작") query = "SELECT COUNT(*) FROM stock_prices" result = execute_query(query) count = result[0][0] if result else 0 logging.info(f"처리된 레코드 수 : ${count}") return {"record_count": count} trigger_dag = DAG( 'trigger_example_dag', default_args=default_args, description='TriggerDagRunOperator 예제', schedule='@daily', catchup=False, tags=['cross-dag', 'trigger'] ) process_task = PythonOperator( task_id='process_data', python_callable=process_data, dag=trigger_dag ) trigger_next_dag = TriggerDagRunOperator( task_id='trigger_sensor_dag', trigger_dag_id='sensor_example_dag', wait_for_completion=False, # 완료를 기다리지 않음 dag=trigger_dag ) process_task >> trigger_next_dag # ======================================================== # 두 번째 DAG: 외부 태스크 대기 def analyze_triggered_data(**context): """트리거된 후 분석 작업""" logging.info("외부 태스크 완료 후 분석 시작") analysis_result = { "status": "completed", "timestamp": str(datetime.now()) } logging.info(f"분석 완료: {analysis_result}") return analysis_result sensor_dag = DAG( 'sensor_example_dag', default_args=default_args, description='ExternalTaskSensor 예제', schedule=None, # 트리거로만 실행됨 catchup=False, tags=['cross-dag', 'sensor'] ) # 외부 DAG의 태스크 완료를 기다림 def get_most_recent_dag_run(dt): """가장 최근의 DAG 실행을 찾는 함수""" from airflow.models import DagRun from airflow.utils.session import provide_session @provide_session def get_recent_run(session=None): recent_run = session.query(DagRun).filter( DagRun.dag_id == 'trigger_example_dag', DagRun.state == 'success' ).order_by(DagRun.execution_date.desc()).first() if recent_run: return recent_run.execution_date return dt return get_recent_run() wait_for_external_task = ExternalTaskSensor( task_id='wait_for_process_data', external_dag_id='trigger_example_dag', # 기다릴 DAG external_task_id='process_data', # 기다릴 태스크 execution_date_fn=get_most_recent_dag_run, timeout=60, # 1분 타임아웃 poke_interval=5, # 5초마다 확인 allowed_states=['success'], dag=sensor_dag ) analyze_task = PythonOperator( task_id='analyze_triggered_data', python_callable=analyze_triggered_data, dag=sensor_dag ) wait_for_external_task >> analyze_task

안녕하세요 Header.jsx 수정중 궁금한점이 있어서 질문드립니다!new Date() 함수는 날짜를 가져오는 함수라 리렌더링이 계속 될것같은데 memo 를 사용하면 리렌더링이 안되는건가요 ??  

안녕하세요 맛비님, 언제나 좋은 내용 감사합니다. 아직 FPGA 수강등록을 하지 않았기에 부득이하게 Verilog 학습 q&a에 질문을 드립니다. 설계독학맛비's 실전 FPGA를 이용한 HW 가속기 설계, 강의를 위해 보드를 구매하려 하는데 Zybo z7-20 board가 단종이 되어 해외에서 또는 중고로 구매해야 하는 약간의 Risk가 있습니다. 금액도 생각보다 좀 나가네요. Arty z7 10으로 시작을 하고, 나중에 좀 실력이 생기면 조금 더 비싼 것으로 구매하던지 하는 Safe한 Approach를 하려 하는데요. Arty board를 구매했을 때 수업 내용과 달라서 어려운 점이 예상 되는 부분이 있나요? Board 구매를 한 후 수강을 하려 생각하고 있습니다. 감사합니다.

지금 깃허브도 만들었고 PS D:\programming\java_kotlin\Java_Spring_Library_Application\library-app\library-app> git push --set-upstream origin masterremote: Invalid username or token. Password authentication is not supported for Git operations.fatal: Authentication failed for 'https://github.com/ej123123123/library-app.git/'이렇게 뜨면서 깃허브 push가 안되는데 왜그럴까요

클러스터링 에이전트까지 진행을 했는데요. 익스큐트 시키면 클러스터링에서 멈추지 않고 계속 돌아가네요. 하도 오래동안 돌아가서 멈췄더니 저렇게 숫자가 보이구요. 혹시나해서 다시 진행해봐도 또 계속 클러스터링에서 돌아갑니다. 뭐가 문제일까요?

Trace Batch Finalization 이 부분이 출력되지않는데요.뭔가 설정이 필요한가요?

 강사님 저는 주제와 대본을 각각 생성하고 싶어서 앞서 말씀 주신 방법대로 셋팅했는데 위와 같이 대본은 아래 줄에 생성이 되어서 혹시 이부분 어떻게 해결하나요?

강의 정보와 관련하여 질문하고 싶습니다.고객센터나 1:1 채팅이 따로 가능한 방법이 있을까요?

학습 관련 질문이 있으시면, 상세하게 남겨주세요.문제가 발생한 부분의 코드를 함께 올려주세요.수업 영상 몇 분/초 구간인지 알려주세요.

안녕하세요. 현재 섹션4까지 학습하고 퀴즈를 풀었는데, 퀴즈에 파일명이 잘못되어 있어 제보합니다. CLAUDE.md 인데 cloud.md 라고 퀴즈에 나오네요. (아래 캡쳐된 2개의 문항에 이렇게 나옵니다)  학습 자체에 큰 지장은 없지만, 그래도 알려드립니다.   

제가 잘못구현한건지 같은 책 이름으로여러개의 책 이름이 들어가는 거 같은데이건 상관없나요?책등록은 여러개 되는데 그에 따라서 대출은 하나밖에 안되는데예를 들면 클린코드를 3개 등록할 수 있는데, 3개 등록해도 대출은 1개밖에 안돼요

안녕하세요, 처음 AWS에 대해 알게 되고 해당 자격증에 대해 알아보다가 이 강의를 듣게 되었었습니다. 수강 이후에 계속 어떻게든 들어보려고 했지만 다른 여러 이슈들을 처리하다보니 어느덧 내일이 수강 마지막 날이 되었습니다; 염치 불구하고 수강 기간연장을 부탁드리며 이번엔 꾸준히 수강하여 유의미한 결과를 낼 수 있도록 하겠습니다. 미리 메리 크리스마스 되시고 새해 복 많이 받으세요.

평생교육 이용권 사용자 입니다2개 강좌를 신청 했는데 바우처 사용내역에 반영이 안되고 있습니다기관에 연락하니 인프런으로 문의 하라고 합니다채움 카드로 결재 했는데 카드 사용내역에는 나오고 바우처 사용내역에는 나오지 않아서 어떻게 해야 바우처 사용으로 확인 할수 있는지 알려주시기 바랍니다

학습을 진행하려고 합니다. 샘플로 제공되어 있는 강의는 영상 및 음성, 자막이 모두 정상적으로 출력이 되는데, 본 강의에서는 영상이 보이지 않습니다. 음성 및 자막은 출력이 됩니다. chrome, firefox에서 테스트 해보았으나 동일한 현상이 나타납니다.

강의 감사합니다. 강의자님의 포트폴리오를 보면서 얼마나 많이 노력하셨는지가 보입니다--공유해주신 포트폴리오를 제 상황에 맞게 하려고 하는데요 ! 질문이 있습니다. 저는 대회를 나간 경력이없고 신입이라 그런지 공유해주신 포폴에 무엇을 쓰면 좋을지 조언듣고 싶습니다. 음 제 지금 상황은 비전공자이고,, react.js 블로그 포폴 하나 혼자 강의 보면서 만들었는데 어떻게 어필하면 좋을까요 .. 자기소개서는 포트폴리오에 올리지 않는게 좋을까요 ? 방향잡기가 쉽지가 않네요

안녕하세요 강사님, 현업 중 궁금한 점이 생겨 질문 드립니다. spacing은 항상 4 혹은 8의 배수 로 진행한다면, 버튼 컴포넌트 생성시에도 동일하게 적용이 되는지 궁금합니다. 강사님의 강의를 접하기전에는, 프론트엔드 개발자 분이 무조건 짝수로 맞춰달라는 요청에패딩을 짝수로 맞췄습니다. 그런데.... 본문에만 적용해야할 행간을 버튼 컴포넌트 폰트에도 동일하게 적용하여 결국엔 버튼 컴포넌트의 최종 크기가 짝수가 아니게 되어버렸습니다. ㅜㅜ결국 개발자 분이 전부 수습을 하셨고, 이런 계기로 강사님의 강의를 듣게 되었는데요.제 이야기의 문제점은1. 버튼 컴포넌트 폰트에 행간 적용 2. 버튼 컴포넌트 패딩 값에 짝수 적용 이라고 생각되어 지는데, 버튼 컴포넌트 전체 너비와 높이가 짝수가 된다면 패딩 값은 홀수로 지정해도 되는것인지 궁금합니다. ㅜㅜ

안녕하세요 선생님! 강의를 듣던 중 static 키워드가 메모리에 올라가는 시점을 확실히 알고 싶어 질문드리게 되었습니다. 강의에서 static은 선언하자마자 메모리에 올라간다고 하셨는데요, 그게 클래스 내부에서 static 변수나 함수를 선언했을 때 메모리에 올라간다는 의미가 맞나요?그러면 100개의 클래스를 작성하고, 거기에 static으로 함수나 변수를 써놓으면 그게 다 사용되지 않아도 메모리에 올라가는 건가요??  

형 코드 보고 있는데 아래와 같이 public static class로 되어 있어. 이게 무슨 의미야 ? 따로 static 메소드는 없는거 같은데  

사진처럼 binding = 할때 빨간색이 사라지지 않습니다. 에러가 나는거 같은데 뭐가 문제가 될까요?2번 사진에서 혹시 뭐 클릭해야하는게 있을까요?

FE에서 모킹을 포함한 통합 테스트를 진행할 때, API 요청 전까지의 동작API 요청API 응답API 응답에 따른 화면 변환 1, 4는 테스트에서 주로 다루고 있고,3은 모킹으로 처리하고 있는데, 2번의 경우, 예를 들어 화면에서 필터로 { name: "apple", limit: 10 } 를 설정하고 요청이 발생했으면, request의 parameter에 이런 요청이 도달했다는 것 까지 검증하면 좋을지가 궁금합니다. 생성/삭제/수정과 같은 API가 포함된 경우 동일 GET API라도 다른 응답이 반환되어야 되는 경우가 있어, 해당 테스트 시 응답 자체를 override해서 모킹하여서 작성하였는데, 이 경우 사실상 API 응답과 API 응답에 따른 화면 변환은 테스트가 성공하는 채로 고정하는데, 실제 요청이 올바르게 이루어졌는지는 검증하지 못하는 것 같아서요.