@task만 쓰거나 inner 함수에 return을 하거나 multiple_outputs=True 옵션(이건 안해봄) 으로 해결되는 것 같습니다. 로그는 다음과 같습니다. Timestamp: 2025-11-01 19:54:50 Traceback (most recent call last): File "<attrs generated methods airflow.sdk.bases.decorator._TaskDecorator>", line 37, in init _setattr('multiple_outputs', __attr_factory_multiple_outputs(self)) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/bases/decorator.py", line 328, in _infer_multiple_outputs if "return" not in self.function.annotations: ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ AttributeError: 'str' object has no attribute 'annotations'. Did you mean: 'contains'?
안녕하십니까, 강사님. PySpark에 대한 심도 있는 강의를 준비해주셔서 감사합니다. PySpark를 공부하기 전에 한 가지 궁금한 점이 있습니다. 많은 분들이 PySpark를 Databricks 플랫폼에서 실행하던데, 특별한 이유가 있을까요? 로컬 환경에 패키지를 설치해 직접 실행하지 않는 이유가 궁금합니다.
https://smithery.ai/server/@smithery-ai/filesystem 위 사이트의 파일시스템 연결하는 부분이 강의랑 좀 달라졌습니다. 경로를 Cursor에 직접 등록하는 부분이 아니라 아래의 Connect 부분에서 Configure 부분에 디렉토리를 입력하도록 변경됬네요. 우측 상단에 Configure 누르시면 allowedDirectories를 입력하는 창이 생깁니다. 주의! 리스트 형태로 넣어줘야 합니다. ["경로1", "경로2"] 처음에는 경로를 하나씩 넣고 Add Item으로 추가해줬는데.. 계속 빨간불 떠서 이것 저것 다해보다가 저 위의 방식으로 해결됬습니다. 참고들 하세요..
- 학습 관련 질문을 남겨주세요. 상세히 작성하면 더 좋아요! - 먼저 유사한 질문이 있었는지 검색해보세요. - 서로 예의를 지키며 존중하는 문화를 만들어가요. - 잠깐! 인프런 서비스 운영 관련 문의는 1:1 문의하기를 이용해주세요. select coalesce ( to_char ( b . order_date , 'yyyy' ), '총매출' ) as year , coalesce ( to_char ( b . order_date , 'mm' ), '연매출' ) as month , coalesce ( to_char ( b . order_date , 'dd' ), '월매출' ) as day , sum ( a . amount ) as sum_amount from nw . order_items a join nw . orders b on a . order_id = b . order_id group by rollup ( to_char ( b . order_date , 'yyyy' ), to_char ( b . order_date , 'mm' ), to_char ( b . order_date , 'dd' )) order by 1 , 2 , 3 ; case when 안쓰고 coalesce 해도 괜찮을것 같아요!
Queue에 대한 질문이 몇가지가 있습니다. 큐를 지정할 때는 무조건 큐의 이름을 지정해 줘야 하나요? 자동으로 비어 있는 woker에 큐를 할당하는 방법은 없나요? cpu_intensive라는 woker에 여러개의 큐가 동시에 요청이 왔을 경우 동기적으로 처리하나요? 큐를 생성하면 해당 큐의 물리적 자원은 어떻게 할당 되는 것인가요? 대체로 하나의 DAG에서 강의에 예시와 같이 여러 개의 큐를 사용하는 경우가 있을까요? 워커를 많이 만들어 환경을 구성하는 사례는 어떤 사례가 있는지 알 수 있을까요? 내용 확인 부탁드립니다.
안녕하세요 빅디님 ! gcc 설치 중에 오류가 나서 yum repository 삭제 후 다시 시도해 보았는데, 계속 오류가 나서 질문 드립니다. ㅠㅠ 어떤게 문제일까요..? yum repo 삭제는 다음과 같이 진행 하였습니다. [root@server02 ~]# cd /etc/yum.repos.d/ [root@server02 yum.repos.d]# rm -rf remi.* remi-* [root@server02 yum.repos.d]# [root@server02 yum.repos.d]# cd /var/cache/yum/ [root@server02 yum]# rm -rf x86_64 [root@server02 yum]# [root@server02 yum]# yum clean headers Loaded plugins: fastestmirror, refresh-packagekit, security Cleaning repos: base cloudera-manager extras updates 0 header files removed [root@server02 yum]# yum clean packages Loaded plugins: fastestmirror, refresh-packagekit, security Cleaning repos: base cloudera-manager extras updates 0 package files removed [root@server02 yum]# yum clean metadata Loaded plugins: fastestmirror, refresh-packagekit, security Cleaning repos: base cloudera-manager extras updates 0 metadata files removed 0 sqlite files removed 0 metadata files removed yum install -y gcc* 명령어 입력시 발생하는 오류 입니다. [root@server02 ~]# yum install -y gcc* Loaded plugins: fastestmirror, refresh-packagekit, security Setting up Install Process Loading mirror speeds from cached hostfile Could not retrieve mirrorlist http://mirrorlist.centos.org/?release=6&arch=x86_64&repo=os&infra=stock error was 14: PYCURL ERROR 6 - "Couldn't resolve host 'mirrorlist.centos.org'" Error: Cannot find a valid baseurl for repo: base 추가로, CentOS-Base.repo 파일 내용 첨부드립니다. [root@server02 yum.repos.d]# cat CentOS-Base.repo # CentOS-Base.repo # # The mirror system uses the connecting IP address of the client and the # update status of each mirror to pick mirrors that are updated to and # geographically close to the client. You should use this for CentOS updates # unless you are manually picking other mirrors. # # If the mirrorlist= does not work for you, as a fall back you can try the # remarked out baseurl= line instead. # # [base] name=CentOS-$releasever - Base mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=os&infra=$infra #baseurl=http://mirror.centos.org/centos/$releasever/os/$basearch/ gpgcheck=1 gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-6 #released updates [updates] name=CentOS-$releasever - Updates mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=updates&infra=$infra #baseurl=http://mirror.centos.org/centos/$releasever/updates/$basearch/ gpgcheck=1 gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-6 #additional packages that may be useful [extras] name=CentOS-$releasever - Extras mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=extras&infra=$infra #baseurl=http://mirror.centos.org/centos/$releasever/extras/$basearch/ gpgcheck=1 gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-6 #additional packages that extend functionality of existing packages [centosplus] name=CentOS-$releasever - Plus mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=centosplus&infra=$infra #baseurl=http://mirror.centos.org/centos/$releasever/centosplus/$basearch/ gpgcheck=1 enabled=0 gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-6 #contrib - packages by Centos Users [contrib] name=CentOS-$releasever - Contrib mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=contrib&infra=$infra #baseurl=http://mirror.centos.org/centos/$releasever/contrib/$basearch/ gpgcheck=1 enabled=0 gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-6 감사합니다.
메시지 키가 있을 경우 전달되는 파티션이 변경되는 것을 방지하기 위해 충분히 큰 파티션으로 생성 을 해야한다. 해당 부분에 대해 궁금한 점이 생겨 질문올립니다. 여기서의 충분히 큰 파티션은 어떻게 산정을 하면 좋을까요? Spring Application으로 Consumer를 구현한다고 했을 때, 충분히 큰 파티션 생성 시 Consumer : Kafka Topic = 1 : N 구조로 구독을 하는 방향으로 만들면 되는 건 지 궁금합니다.
안녕하세요 김현진 선생님 강의 잘 듣고 있습니다. 아래의 퀴즈를 풀다가 궁금한 점이 생겨서 질문을 남깁니다. 스케줄러의 DAG 파일 파싱 부하를 줄이기 위해 권장되는 코드 작성 방법은 무엇일까요? --> 라이브러리 import 문을 Python callable 함수 내부로 옮긴다. 위 설명을 이해하기 위해서 상황을 가정해보았는데요, python callable 함수 외부에 선언된 API 호출문이 있는 경우를 피해야하고, 만약 그렇게 되면 스케쥴러가 DAG을 파싱할 때마다 실행되는 상황이 되는 것이 맞을까요?
이 부분도 이해가 잘 안 가서 추가로 질문드려요. Sink 과정을 자세히 쪼개보면 SinkWriter, Committer, Global Committer로 나눌 수 있는데, SinkWriter로 로컬에 데이터를 임시로 쓰면 Committer가 커밋을 수행한다는 내용이 맞나요? 그리고 Topology 용어도 궁금한데요 Topology가 네트워크 시간에 노드들을 연결해놓은 방식이라고 배웠었는데요 여기서 Topology가 어떤 뜻으로 사용되나요? (강의 덕분에 단어도 많이 알아가네요) 마지막으로 flink 문서를 찾아보니 SinkWriter, Committer, Global Committer 클래스가 삭제 되었다고 나오는데요, 버전업이 되면서 이제는 이런 방식으로 동작하지 않는 건가요? 답변 미리 감사드립니다.. 🙌 저는 오늘 이까지 들으려구요. 휴가 잘 다녀오세요!!!
안녕하세요 강사님, 강의 중 이 페이지가 도저히 이해가 안 돼서 추가적으로 (되도록이면 한글로) 풀어서 설명 부탁드려도 될까요? 질문드릴 내용과 제가 이해한게 맞는지 확인하는 부분 두 부분으로 나눠서 작성해두었습니다. [질문드릴 내용] First Class Support에 대해서 궁금합니다. 왜 First Class라는 용어를 사용하나요? Periodic checkpointing, Robust의 차이가 궁금합니다. 유추상.. Periodic checkpointing은 주기적으로 체크포인트를 지정해서 체크포인트 기준으로 다시 동작시키기 때문에 중복 처리를 할 수 있는데, Flink는 exactly once를 지원하기 때문에 무조건 한 번만 실행함을 보장하는 건가요? Backpressure는 데이터가 많이 들어와 병목이 생길 때 처리인데, Spark는 지원 범위가 좁고 Flink는 세밀하게 지원 가능한건가요? 추가로 fine-grained 용어가 무슨 뜻인지도 궁금합니다. [제가 이해한게 맞는지 확인 부탁드려요] maintain state across events: flink가 event 기반으로 동작하고, 이벤트간의 상태를 알고 있다는 뜻인가요? Spark이 Standalone하고 윈도우 사이에 연결이 없다: 추측상 데이터를 윈도우 단위로 잘라서 처리하고, 윈도우끼리 상태를 공유하지 않는다는 뜻 맞을까요? [설명 스크립트] Apache Flinke Better Fault Tolerancee-Consistency를 가지고요. Across Events 간의 maintain 합니다. 예를 들어서 유저 간의 세션 데이터를 다 계속 모아서 사용할 수가 있어요. 그에 반해서 Spark는 심플하고요. Standalone, 윈도우 간에 연결하는 게 없고요. Light, Stateful, Workload라고 생각하시면 됩니다. 스파크는 스테이트풀 스트림을 베이직 서포트만 한다 고 하는 거고요. Flink는 Across 이벤트별로 스테이트를 메인테인하기 때문에 First Class Support로 시작할 때부터 그런 식으로 만들었습니다. 체크포인트 측면에서도 Spark는 Periodically 체크 포인트를 하지만 Flink는 Robust, Exactly Once 체크포인트를 지원해서 아 주 작은,very low한 오버헤드를 가지고 있습니다 Backpressure 같은 경우는 데이터가 엄청나게 많이 들어오잖아요 그럴 경우에 어떻게 핸들할 건가 기본적인 것만 스파크는 서 포트를 하구요 fine-grained 하지않습니다. 그에 반해서 이제 Flink 같은 경우에는 백프레셔 메카니즘이 기본적으로 빌트인되어있기 때문에 Smooth flow control과 stability가 같이 들어 있습니다.
안녕하세요, 강사님이 실무에서 dbt를 사용하시면서 느끼신 단점 혹은 한계점이 어떤 게 있는지 궁금합니다. 예시로, connection은 너무나 많고, 각각의 구현에 의존하다 보니 생기는 불편한 점이 종종 있습니다. GCP BigQuery의 경우 custom labeling이 dataset 단위는 불가능, table 단위로만 가능하기 때문에 dataset 단위로 cost가 측정되는 GCP 특정 상 custom label 단위로 cost breakdown이 어려워진다는 한계점이 있습니다. (as of 2025.10)
